feat: cosmovisor batch upgrades

tools/cosmovisor/args.go

@@ -109,6 +109,10 @@ func (cfg *Config) UpgradeInfoFilePath() string {
 	return filepath.Join(cfg.Home, "data", upgradetypes.UpgradeInfoFilename)
 }
 
+func (cfg *Config) UpgradeInfoBatchFilePath() string {
+	return cfg.UpgradeInfoFilePath() + ".batch"
+}
+
 // SymLinkToGenesis creates a symbolic link from "./current" to the genesis directory.
 func (cfg *Config) SymLinkToGenesis() (string, error) {
 	genesis := filepath.Join(cfg.Root(), genesisDir)

tools/cosmovisor/cmd/cosmovisor/add_upgrade.go

@@ -19,7 +19,7 @@ func NewAddUpgradeCmd() *cobra.Command {
 		Short:        "Add APP upgrade binary to cosmovisor",
 		SilenceUsage: true,
 		Args:         cobra.ExactArgs(2),
-		RunE:         AddUpgrade,
+		RunE:         AddUpgradeCmd,
 	}
 
 	addUpgrade.Flags().Bool(cosmovisor.FlagForce, false, "overwrite existing upgrade binary / upgrade-info.json file")
@@ -28,26 +28,13 @@ func NewAddUpgradeCmd() *cobra.Command {
 	return addUpgrade
 }
 
-// AddUpgrade adds upgrade info to manifest
-func AddUpgrade(cmd *cobra.Command, args []string) error {
-	configPath, err := cmd.Flags().GetString(cosmovisor.FlagCosmovisorConfig)
-	if err != nil {
-		return fmt.Errorf("failed to get config flag: %w", err)
-	}
-
-	cfg, err := cosmovisor.GetConfigFromFile(configPath)
-	if err != nil {
-		return err
-	}
-
+func AddUpgrade(cfg *cosmovisor.Config, force bool, upgradeHeight int64, upgradeName, executablePath, upgradeInfoPath string) error {
 	logger := cfg.Logger(os.Stdout)
 
-	upgradeName := args[0]
 	if !cfg.DisableRecase {
-		upgradeName = strings.ToLower(args[0])
+		upgradeName = strings.ToLower(upgradeName)
 	}
 
-	executablePath := args[1]
 	if _, err := os.Stat(executablePath); err != nil {
 		if os.IsNotExist(err) {
 			return fmt.Errorf("invalid executable path: %w", err)
@@ -68,21 +55,14 @@ func AddUpgrade(cmd *cobra.Command, args []string) error {
 		return fmt.Errorf("failed to read binary: %w", err)
 	}
 
-	force, err := cmd.Flags().GetBool(cosmovisor.FlagForce)
-	if err != nil {
-		return fmt.Errorf("failed to get force flag: %w", err)
-	}
-
 	if err := saveOrAbort(cfg.UpgradeBin(upgradeName), executableData, force); err != nil {
 		return err
 	}
 
 	logger.Info(fmt.Sprintf("Using %s for %s upgrade", executablePath, upgradeName))
 	logger.Info(fmt.Sprintf("Upgrade binary located at %s", cfg.UpgradeBin(upgradeName)))
 
-	if upgradeHeight, err := cmd.Flags().GetInt64(cosmovisor.FlagUpgradeHeight); err != nil {
-		return fmt.Errorf("failed to get upgrade-height flag: %w", err)
-	} else if upgradeHeight > 0 {
+	if upgradeHeight > 0 {
 		plan := upgradetypes.Plan{Name: upgradeName, Height: upgradeHeight}
 		if err := plan.ValidateBasic(); err != nil {
 			panic(fmt.Errorf("something is wrong with cosmovisor: %w", err))
@@ -104,6 +84,41 @@ func AddUpgrade(cmd *cobra.Command, args []string) error {
 	return nil
 }
 
+func GetConfig(cmd *cobra.Command) (*cosmovisor.Config, error) {
+	configPath, err := cmd.Flags().GetString(cosmovisor.FlagCosmovisorConfig)
+	if err != nil {
+		return nil, fmt.Errorf("failed to get config flag: %w", err)
+	}
+
+	cfg, err := cosmovisor.GetConfigFromFile(configPath)
+	if err != nil {
+		return nil, err
+	}
+	return cfg, nil
+}
+
+// AddUpgrade adds upgrade info to manifest
+func AddUpgradeCmd(cmd *cobra.Command, args []string) error {
+	cfg, err := GetConfig(cmd)
+	if err != nil {
+		return err
+	}
+
+	upgradeName, executablePath := args[0], args[1]
+
+	force, err := cmd.Flags().GetBool(cosmovisor.FlagForce)
+	if err != nil {
+		return fmt.Errorf("failed to get force flag: %w", err)
+	}
+
+	upgradeHeight, err := cmd.Flags().GetInt64(cosmovisor.FlagUpgradeHeight)
+	if err != nil {
+		return fmt.Errorf("failed to get upgrade-height flag: %w", err)
+	}
+
+	return AddUpgrade(cfg, force, upgradeHeight, upgradeName, executablePath, cfg.UpgradeInfoFilePath())
+}
+
 // saveOrAbort saves data to path or aborts if file exists and force is false
 func saveOrAbort(path string, data []byte, force bool) error {
 	if _, err := os.Stat(path); err == nil {

tools/cosmovisor/cmd/cosmovisor/batch_upgrade.go

@@ -0,0 +1,77 @@
+package main
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"strconv"
+	"strings"
+
+	"github.com/spf13/cobra"
+)
+
+func NewBatchAddUpgradeCmd() *cobra.Command {
+	return &cobra.Command{
+		Use:          "add-batch-upgrade <upgrade1-name>:<path-to-exec1>:<upgrade1-height> <upgrade2-name>:<path-to-exec2>:<upgrade2-height> .. <upgradeN-name>:<path-to-execN>:<upgradeN-height>",
+		Short:        "Add APP upgrades binary to cosmovisor",
+		SilenceUsage: true,
+		Args:         cobra.ArbitraryArgs,
+		RunE:         AddBatchUpgrade,
+	}
+}
+
+func AddBatchUpgrade(cmd *cobra.Command, args []string) error {
+	cfg, err := GetConfig(cmd)
+	if err != nil {
+		return err
+	}
+	upgradeInfoPaths := []string{}
+	for i, as := range args {
+		a := strings.Split(as, ":")
+		if len(a) != 3 {
+			return fmt.Errorf("argument at position %d (%s) is invalid", i, as)
+		}
+		upgradeName := args[0]
+		upgradePath := args[1]
+		upgradeHeight, err := strconv.ParseInt(args[2], 10, 64)
+		if err != nil {
+			return fmt.Errorf("upgrade height at position %d (%s) is invalid", i, args[2])
+		}
+		upgradeInfoPath := cfg.UpgradeInfoFilePath() + upgradeName
+		upgradeInfoPaths = append(upgradeInfoPaths, upgradeInfoPath)
+		if err := AddUpgrade(cfg, true, upgradeHeight, upgradeName, upgradePath, upgradeInfoPath); err != nil {
+			return err
+		}
+	}
+
+	var allData []json.RawMessage
+	for _, uip := range upgradeInfoPaths {
+		fileData, err := os.ReadFile(uip)
+		if err != nil {
+			return fmt.Errorf("Error reading file %s: %v", uip, err)
+		}
+
+		// Verify it's valid JSON
+		var jsonData json.RawMessage
+		if err := json.Unmarshal(fileData, &jsonData); err != nil {
+			return fmt.Errorf("Error parsing JSON from file %s: %v", uip, err)
+		}
+
+		// Add to our slice
+		allData = append(allData, jsonData)
+	}
+
+	// Marshal the combined data
+	batchData, err := json.MarshalIndent(allData, "", "  ")
+	if err != nil {
+		return fmt.Errorf("Error marshaling combined JSON: %v", err)
+	}
+
+	// Write to output file
+	err = os.WriteFile(cfg.UpgradeInfoBatchFilePath(), batchData, 0644)
+	if err != nil {
+		return fmt.Errorf("Error writing combined JSON to file: %v", err)
+	}
+
+	return nil
+}

tools/cosmovisor/process.go

@@ -1,6 +1,7 @@
 package cosmovisor
 
 import (
+	"context"
 	"encoding/json"
 	"errors"
 	"fmt"
@@ -9,6 +10,7 @@
 	"os/exec"
 	"os/signal"
 	"path/filepath"
+	"sort"
 	"strconv"
 	"strings"
 	"syscall"
@@ -19,6 +21,8 @@
 	"cosmossdk.io/log"
 	"cosmossdk.io/x/upgrade/plan"
 	upgradetypes "cosmossdk.io/x/upgrade/types"
+	"github.com/cometbft/cometbft/rpc/client/http"
+	cmttypes "github.com/cometbft/cometbft/types"
 )
 
 type Launcher struct {
@@ -36,6 +40,71 @@
 	return Launcher{logger: logger, cfg: cfg, fw: fw}, nil
 }
 
+func BatchWatcher(ctx context.Context, cfg *Config, logger log.Logger, rpcAddress string) error {
+	// load batch file in memory
+	var uInfos []upgradetypes.Plan
+	upgradeInfoFile, err := os.ReadFile(cfg.UpgradeInfoBatchFilePath())
+	if err != nil {
+		return fmt.Errorf("error while reading upgrade-info.json.batch: %w", err)
+	}
+
+	if err = json.Unmarshal(upgradeInfoFile, &uInfos); err != nil {
+		return err
+	}
+	sort.Slice(uInfos, func(i, j int) bool {
+		return uInfos[i].Height < uInfos[j].Height
+	})
+
+	client, err := http.New(rpcAddress, "/websocket")
+	if err != nil {
+		return fmt.Errorf("failed to create CometBFT client: %w", err)
+	}
+	defer client.Stop()
+
+	err = client.Start()
+	if err != nil {
+		return fmt.Errorf("failed to start CometBFT client: %w", err)
+	}
+
+	eventCh, err := client.Subscribe(ctx, "cosmovisor-watcher", cmttypes.EventQueryNewBlock.String())
+	if err != nil {
+		return fmt.Errorf("failed to subscribe to new blocks: %w", err)
+	}
+	// TODO: break if empty array
+	for {
+		select {
+		case e := <-eventCh:
+			nb, ok := e.Data.(cmttypes.EventDataNewBlock)
+			if !ok {
+				logger.Warn("batch watcher: unexpected event data type:", e.Data)
+				continue
+			}
+			h := nb.Block.Height
+			// read batch file in buffer, get lowest, compare, replace and delete
+			if h > uInfos[0].Height {
+				jsonBytes, err := json.Marshal(uInfos[0])
+				if err != nil {
+					return fmt.Errorf("error marshaling JSON: %w", err)
+				}
+				if err := os.WriteFile(cfg.UpgradeInfoFilePath(), jsonBytes, 0o755); err != nil {
+					return fmt.Errorf("error writing upgrade-info.json: %w", err)
+				}
+				uInfos = uInfos[1:]
+
+				jsonBytes, err = json.Marshal(uInfos)
+				if err != nil {
+					return fmt.Errorf("error marshaling JSON: %w", err)
+				}
+				if err := os.WriteFile(cfg.UpgradeInfoBatchFilePath(), jsonBytes, 0o755); err != nil {
+					return fmt.Errorf("error writing upgrade-info.json.batch: %w", err)
+				}
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
 // Run launches the app in a subprocess and returns when the subprocess (app)
 // exits (either when it dies, or *after* a successful upgrade.) and upgrade finished.
 // Returns true if the upgrade request was detected and the upgrade process started.
@@ -58,10 +127,13 @@
 		return false, fmt.Errorf("launching process %s %s failed: %w", bin, strings.Join(args, " "), err)
 	}
 
+	ctx, cancel := context.WithCancel(context.Background())
+
 	sigs := make(chan os.Signal, 1)
 	signal.Notify(sigs, syscall.SIGQUIT, syscall.SIGTERM)
 	go func() {
 		sig := <-sigs
+		cancel()
 		if err := cmd.Process.Signal(sig); err != nil {
 			l.logger.Error("terminated", "error", err, "bin", bin)
 			os.Exit(1)
@@ -72,6 +144,8 @@
 		return false, err
 	}
 
+	go BatchWatcher(ctx, l.cfg, l.logger, "localhost")
+
 	if !IsSkipUpgradeHeight(args, l.fw.currentInfo) {
 		l.cfg.WaitRestartDelay()