feat: add coordinated zetaclient restart

feedback updates

cmd/zetae2e/local/local.go

@@ -362,6 +362,7 @@ func localE2ETest(cmd *cobra.Command, _ []string) {
 
 	if testAdmin {
 		eg.Go(adminTestRoutine(conf, deployerRunner, verbose,
+			e2etests.TestOperationalFlagsName,
 			e2etests.TestWhitelistERC20Name,
 			e2etests.TestPauseZRC20Name,
 			e2etests.TestUpdateBytecodeZRC20Name,

e2e/e2etests/e2etests.go

@@ -128,6 +128,7 @@ const (
 	TestMigrateERC20CustodyFundsName  = "migrate_erc20_custody_funds"
 	TestMigrateTSSName                = "migrate_TSS"
 	TestSolanaWhitelistSPLName        = "solana_whitelist_spl"
+	TestOperationalFlagsName          = "operational_flags"
 
 	/*
 	 Operational tests
@@ -838,7 +839,12 @@ var AllE2ETests = []runner.E2ETest{
 		[]runner.ArgDefinition{},
 		TestMigrateERC20CustodyFunds,
 	),
-
+	runner.NewE2ETest(
+		TestOperationalFlagsName,
+		"operational flags functionality",
+		[]runner.ArgDefinition{},
+		TestOperationalFlags,
+	),
 	/*
 	 Special tests
 	*/

e2e/e2etests/test_operational_flags.go

@@ -0,0 +1,38 @@
+package e2etests
+
+import (
+	"github.com/stretchr/testify/require"
+
+	"github.com/zeta-chain/node/e2e/runner"
+	"github.com/zeta-chain/node/e2e/utils"
+	observertypes "github.com/zeta-chain/node/x/observer/types"
+)
+
+// TestOperationalFlags tests the functionality of operations flags.
+func TestOperationalFlags(r *runner.E2ERunner, _ []string) {
+	operationalFlagsRes, err := r.Clients.Zetacore.Observer.OperationalFlags(
+		r.Ctx,
+		&observertypes.QueryOperationalFlagsRequest{},
+	)
+	require.NoError(r, err)
+
+	// always set to low height so it's ignored by zetaclient
+	nextRestartHeight := operationalFlagsRes.OperationalFlags.RestartHeight + 1
+
+	updateMsg := observertypes.NewMsgUpdateOperationalFlags(
+		r.ZetaTxServer.MustGetAccountAddressFromName(utils.OperationalPolicyName),
+		observertypes.OperationalFlags{
+			RestartHeight: nextRestartHeight,
+		},
+	)
+
+	_, err = r.ZetaTxServer.BroadcastTx(utils.OperationalPolicyName, updateMsg)
+	require.NoError(r, err)
+
+	operationalFlagsRes, err = r.Clients.Zetacore.Observer.OperationalFlags(
+		r.Ctx,
+		&observertypes.QueryOperationalFlagsRequest{},
+	)
+	require.NoError(r, err)
+	require.Equal(r, nextRestartHeight, operationalFlagsRes.OperationalFlags.RestartHeight)
+}

proto/zetachain/zetacore/observer/operational.proto

@@ -0,0 +1,10 @@
+syntax = "proto3";
+package zetachain.zetacore.observer;
+
+option go_package = "github.com/zeta-chain/node/x/observer/types";
+
+message OperationalFlags {
+  // Height for a coordinated zetaclient restart.
+  // Will be ignored if missed.
+  int64 restart_height = 1;
+}

proto/zetachain/zetacore/observer/query.proto

@@ -14,6 +14,7 @@ import "zetachain/zetacore/observer/observer.proto";
 import "zetachain/zetacore/observer/params.proto";
 import "zetachain/zetacore/observer/pending_nonces.proto";
 import "zetachain/zetacore/observer/tss.proto";
+import "zetachain/zetacore/observer/operational.proto";
 import "zetachain/zetacore/pkg/chains/chains.proto";
 import "zetachain/zetacore/pkg/proofs/proofs.proto";
 import "zetachain/zetacore/observer/tss_funds_migrator.proto";
@@ -164,6 +165,18 @@ service Query {
     option (google.api.http).get =
         "/zeta-chain/observer/getAllTssFundsMigrators";
   }
+
+  // Queries operational flags
+  rpc OperationalFlags(QueryOperationalFlagsRequest)
+      returns (QueryOperationalFlagsResponse) {
+    option (google.api.http).get = "/zeta-chain/observer/operationalFlags";
+  }
+}
+
+message QueryOperationalFlagsRequest {}
+
+message QueryOperationalFlagsResponse {
+  OperationalFlags operational_flags = 1 [ (gogoproto.nullable) = false ];
 }
 
 message QueryTssFundsMigratorInfoAllRequest {}

proto/zetachain/zetacore/observer/tx.proto

@@ -8,6 +8,7 @@ import "zetachain/zetacore/observer/observer.proto";
 import "zetachain/zetacore/observer/params.proto";
 import "zetachain/zetacore/observer/pending_nonces.proto";
 import "zetachain/zetacore/observer/tss.proto";
+import "zetachain/zetacore/observer/operational.proto";
 import "zetachain/zetacore/pkg/chains/chains.proto";
 import "zetachain/zetacore/pkg/proofs/proofs.proto";
 
@@ -31,6 +32,8 @@ service Msg {
   rpc DisableCCTX(MsgDisableCCTX) returns (MsgDisableCCTXResponse);
   rpc UpdateGasPriceIncreaseFlags(MsgUpdateGasPriceIncreaseFlags)
       returns (MsgUpdateGasPriceIncreaseFlagsResponse);
+  rpc UpdateOperationalFlags(MsgUpdateOperationalFlags)
+      returns (MsgUpdateOperationalFlagsResponse);
 }
 
 message MsgUpdateObserver {
@@ -136,4 +139,11 @@ message MsgUpdateGasPriceIncreaseFlags {
       [ (gogoproto.nullable) = false ];
 }
 
-message MsgUpdateGasPriceIncreaseFlagsResponse {}
+message MsgUpdateGasPriceIncreaseFlagsResponse {}
+
+message MsgUpdateOperationalFlags {
+  string creator = 1;
+  OperationalFlags operationalFlags = 2 [ (gogoproto.nullable) = false ];
+}
+
+message MsgUpdateOperationalFlagsResponse {}

x/authority/types/authorization_list.go

@@ -20,6 +20,7 @@ var (
 		"/zetachain.zetacore.observer.MsgUpdateChainParams",
 		"/zetachain.zetacore.observer.MsgEnableCCTX",
 		"/zetachain.zetacore.observer.MsgUpdateGasPriceIncreaseFlags",
+		"/zetachain.zetacore.observer.MsgUpdateOperationalFlags",
 	}
 	// AdminPolicyMessages keeps track of the message URLs that can, by default, only be executed by admin policy address
 	AdminPolicyMessages = []string{

x/authority/types/authorization_list_test.go

@@ -399,6 +399,7 @@ func TestDefaultAuthorizationsList(t *testing.T) {
 			sdk.MsgTypeURL(&observertypes.MsgUpdateChainParams{}),
 			sdk.MsgTypeURL(&observertypes.MsgEnableCCTX{}),
 			sdk.MsgTypeURL(&observertypes.MsgUpdateGasPriceIncreaseFlags{}),
+			sdk.MsgTypeURL(&observertypes.MsgUpdateOperationalFlags{}),
 		}
 
 		// EmergencyPolicyMessageList is a list of messages that can be authorized by the emergency policy

x/observer/client/cli/query.go

@@ -43,6 +43,7 @@ func GetQueryCmd(_ string) *cobra.Command {
 		CmdListPendingNonces(),
 		CmdGetAllTssFundsMigrator(),
 		CmdGetTssFundsMigrator(),
+		CmdShowOperationalFlags(),
 	)
 
 	return cmd

x/observer/client/cli/query_operational_flags.go

@@ -0,0 +1,37 @@
+package cli
+
+import (
+	"context"
+
+	"github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/client/flags"
+	"github.com/spf13/cobra"
+
+	"github.com/zeta-chain/node/x/observer/types"
+)
+
+func CmdShowOperationalFlags() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "show-operational-flags",
+		Short: "shows the operational flags",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) error {
+			clientCtx := client.GetClientContextFromCmd(cmd)
+
+			queryClient := types.NewQueryClient(clientCtx)
+
+			params := &types.QueryOperationalFlagsRequest{}
+
+			res, err := queryClient.OperationalFlags(context.Background(), params)
+			if err != nil {
+				return err
+			}
+
+			return clientCtx.PrintProto(res)
+		},
+	}
+
+	flags.AddQueryFlagsToCmd(cmd)
+
+	return cmd
+}

x/observer/client/cli/tx.go

@@ -32,6 +32,7 @@ func GetTxCmd() *cobra.Command {
 		CmdEnableCCTX(),
 		CmdDisableCCTX(),
 		CmdUpdateGasPriceIncreaseFlags(),
+		CmdUpdateOperationalFlags(),
 	)
 
 	return cmd

x/observer/client/cli/tx_update_operational_flags.go

@@ -0,0 +1,64 @@
+package cli
+
+import (
+	"encoding/json"
+	"os"
+
+	"github.com/cosmos/cosmos-sdk/client"
+	"github.com/cosmos/cosmos-sdk/client/flags"
+	"github.com/cosmos/cosmos-sdk/client/tx"
+	"github.com/spf13/cobra"
+
+	"github.com/zeta-chain/node/x/observer/types"
+)
+
+const (
+	fileFlag          = "file"
+	restartHeightFlag = "restart-height"
+)
+
+func CmdUpdateOperationalFlags() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "update-operational-flags",
+		Short: "Broadcast message UpdateOperationalFlags",
+		Args:  cobra.NoArgs,
+		RunE: func(cmd *cobra.Command, _ []string) (err error) {
+			clientCtx, err := client.GetClientTxContext(cmd)
+			if err != nil {
+				return err
+			}
+
+			var operationalFlags types.OperationalFlags
+
+			flagSet := cmd.Flags()
+			file, _ := flagSet.GetString(fileFlag)
+			restartHeight, _ := flagSet.GetInt64(restartHeightFlag)
+
+			if file != "" {
+				input, err := os.ReadFile(file) // #nosec G304
+				if err != nil {
+					return err
+				}
+				err = json.Unmarshal(input, &operationalFlags)
+				if err != nil {
+					return err
+				}
+			} else {
+				operationalFlags.RestartHeight = restartHeight
+			}
+
+			msg := types.NewMsgUpdateOperationalFlags(
+				clientCtx.GetFromAddress().String(),
+				operationalFlags,
+			)
+
+			return tx.GenerateOrBroadcastTxCLI(clientCtx, cmd.Flags(), msg)
+		},
+	}
+
+	cmd.Flags().String(fileFlag, "", "Path to a JSON file containing OperationalFlags")
+	cmd.Flags().Int64(restartHeightFlag, 0, "Height for a coordinated zetaclient restart")
+	flags.AddTxFlagsToCmd(cmd)
+
+	return cmd
+}

x/observer/keeper/grpc_query_operational_flags.go

@@ -0,0 +1,26 @@
+package keeper
+
+import (
+	"context"
+
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	"google.golang.org/grpc/codes"
+	"google.golang.org/grpc/status"
+
+	"github.com/zeta-chain/node/x/observer/types"
+)
+
+func (k Keeper) OperationalFlags(
+	c context.Context,
+	req *types.QueryOperationalFlagsRequest,
+) (*types.QueryOperationalFlagsResponse, error) {
+	if req == nil {
+		return nil, status.Error(codes.InvalidArgument, "invalid request")
+	}
+	ctx := sdk.UnwrapSDKContext(c)
+	// ignoring found is intentional
+	operationalFlags, _ := k.GetOperationalFlags(ctx)
+	return &types.QueryOperationalFlagsResponse{
+		OperationalFlags: operationalFlags,
+	}, nil
+}

x/observer/keeper/msg_server_update_operational_flags.go

@@ -0,0 +1,28 @@
+package keeper
+
+import (
+	"context"
+
+	"cosmossdk.io/errors"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	authoritytypes "github.com/zeta-chain/node/x/authority/types"
+	"github.com/zeta-chain/node/x/observer/types"
+)
+
+func (k msgServer) UpdateOperationalFlags(
+	goCtx context.Context,
+	msg *types.MsgUpdateOperationalFlags,
+) (*types.MsgUpdateOperationalFlagsResponse, error) {
+	ctx := sdk.UnwrapSDKContext(goCtx)
+
+	// check permission
+	err := k.GetAuthorityKeeper().CheckAuthorization(ctx, msg)
+	if err != nil {
+		return nil, errors.Wrap(authoritytypes.ErrUnauthorized, err.Error())
+	}
+
+	k.Keeper.SetOperationalFlags(ctx, msg.OperationalFlags)
+
+	return &types.MsgUpdateOperationalFlagsResponse{}, nil
+}

x/observer/keeper/operational_flags.go

@@ -0,0 +1,26 @@
+package keeper
+
+import (
+	sdk "github.com/cosmos/cosmos-sdk/types"
+
+	"github.com/zeta-chain/node/x/observer/types"
+)
+
+func (k Keeper) SetOperationalFlags(ctx sdk.Context, operationalFlags types.OperationalFlags) {
+	store := ctx.KVStore(k.storeKey)
+	b := k.cdc.MustMarshal(&operationalFlags)
+	key := types.KeyPrefix(types.OperationalFlagsKey)
+	store.Set(key, b)
+}
+
+func (k Keeper) GetOperationalFlags(ctx sdk.Context) (val types.OperationalFlags, found bool) {
+	found = false
+	store := ctx.KVStore(k.storeKey)
+	b := store.Get(types.KeyPrefix(types.OperationalFlagsKey))
+	if b == nil {
+		return
+	}
+	found = true
+	k.cdc.MustUnmarshal(b, &val)
+	return
+}

x/observer/types/codec.go

@@ -20,6 +20,7 @@ func RegisterCodec(cdc *codec.LegacyAmino) {
 	cdc.RegisterConcrete(&MsgEnableCCTX{}, "observer/EnableCCTX", nil)
 	cdc.RegisterConcrete(&MsgDisableCCTX{}, "observer/DisableCCTX", nil)
 	cdc.RegisterConcrete(&MsgUpdateGasPriceIncreaseFlags{}, "observer/UpdateGasPriceIncreaseFlags", nil)
+	cdc.RegisterConcrete(&MsgUpdateOperationalFlags{}, "observer/UpdateOperationalFlags", nil)
 }
 
 func RegisterInterfaces(registry cdctypes.InterfaceRegistry) {
@@ -36,6 +37,7 @@ func RegisterInterfaces(registry cdctypes.InterfaceRegistry) {
 		&MsgEnableCCTX{},
 		&MsgDisableCCTX{},
 		&MsgUpdateGasPriceIncreaseFlags{},
+		&MsgUpdateOperationalFlags{},
 	)
 
 	msgservice.RegisterMsgServiceDesc(registry, &_Msg_serviceDesc)

x/observer/types/keys.go

@@ -87,6 +87,8 @@ const (
 	NonceToCctxKeyPrefix = "NonceToCctx-value-"
 
 	ParamsKey = "Params-value-"
+
+	OperationalFlagsKey = "OperationalFlags-value-"
 )
 
 func GetBlameIndex(chainID int64, nonce uint64, digest string, height uint64) string {