Forking: bank keeper with local-first reads and TLS gRPC fallback; wire in app; scaffold forking KV under x/bloctopus

app/app.go

@@ -18,6 +18,7 @@ import (
 	"cosmossdk.io/client/v2/autocli"
 	"cosmossdk.io/core/appmodule"
 	"cosmossdk.io/log"
+	corestore "cosmossdk.io/core/store"
 	storetypes "cosmossdk.io/store/types"
 	"cosmossdk.io/x/upgrade"
 	upgradekeeper "cosmossdk.io/x/upgrade/keeper"
@@ -66,6 +67,9 @@ import (
 	"github.com/cosmos/cosmos-sdk/x/mint"
 	mintkeeper "github.com/cosmos/cosmos-sdk/x/mint/keeper"
 	minttypes "github.com/cosmos/cosmos-sdk/x/mint/types"
+
+	forkbankkeeper "gitlab.com/thorchain/thornode/v3/x/bloctopus/forkbank/keeper"
+	forking "gitlab.com/thorchain/thornode/v3/x/bloctopus/forking"
 	"github.com/cosmos/cosmos-sdk/x/params"
 	paramskeeper "github.com/cosmos/cosmos-sdk/x/params/keeper"
 	paramstypes "github.com/cosmos/cosmos-sdk/x/params/types"
@@ -147,7 +151,7 @@ type THORChainApp struct {
 	// keepers
 	AccountKeeper authkeeper.AccountKeeper
 	AuthzKeeper   authzkeeper.Keeper
-	BankKeeper    bankkeeper.BaseKeeper
+	BankKeeper    bankkeeper.Keeper
 	StakingKeeper *stakingkeeper.Keeper
 	MintKeeper    mintkeeper.Keeper
 	UpgradeKeeper *upgradekeeper.Keeper
@@ -305,14 +309,37 @@ func NewChainApp(
 		app.MsgServiceRouter(),
 		app.AccountKeeper,
 	)
-	app.BankKeeper = bankkeeper.NewBaseKeeper(
+	var bankStoreSvc corestore.KVStoreService
+	if cast.ToBool(appOpts.Get("fork.enabled")) {
+		bankStoreSvc = forking.NewKVStoreService(runtime.NewKVStoreService(keys[banktypes.StoreKey]))
+	} else {
+		bankStoreSvc = runtime.NewKVStoreService(keys[banktypes.StoreKey])
+	}
+	baseBank := bankkeeper.NewBaseKeeper(
 		app.appCodec,
-		runtime.NewKVStoreService(keys[banktypes.StoreKey]),
+		bankStoreSvc,
 		app.AccountKeeper,
 		BlockedAddresses(),
 		authtypes.NewModuleAddress(thorchain.ModuleName).String(),
 		logger,
 	)
+	app.BankKeeper = baseBank
+	if cast.ToBool(appOpts.Get("fork.enabled")) {
+		endpoint := cast.ToString(appOpts.Get("fork.grpc"))
+		if endpoint == "" {
+			endpoint = "grpc.thor.pfc.zone:443"
+		}
+		fbk, err := forkbankkeeper.NewForkingBankKeeper(baseBank, forkbankkeeper.Config{
+			Endpoint: endpoint,
+		})
+		if err != nil {
+			panic(err)
+		}
+		sdkCtx := app.BaseApp.NewUncachedContext(false, tmproto.Header{})
+		fbk.EnsureDenomMetadata(sdkCtx)
+
+		app.BankKeeper = &fbk
+	}
 
 	txSigningOptions, err := tx.NewDefaultSigningOptions()
 	if err != nil {

cmd/thornode/commands.go

@@ -154,13 +154,10 @@ func addModuleInitFlags(startCmd *cobra.Command) {
 	startCmd.PreRunE = func(cmd *cobra.Command, _ []string) error {
 		serverCtx := server.GetServerContextFromCmd(cmd)
 
-		// Bind flags to the Context's Viper so the app construction can set
-		// options accordingly.
 		if err := serverCtx.Viper.BindPFlags(cmd.Flags()); err != nil {
 			return fmt.Errorf("fail to bind flags,err: %w", err)
 		}
 
-		// replace sdk logger with thorlog
 		if zl, ok := serverCtx.Logger.Impl().(*zerolog.Logger); ok {
 			logger := zl.With().CallerWithSkipFrameCount(3).Logger()
 			serverCtx.Logger = thorlog.SdkLogWrapper{
@@ -172,6 +169,19 @@ func addModuleInitFlags(startCmd *cobra.Command) {
 	}
 	wasm.AddModuleInitFlags(startCmd)
 	ebifrost.AddModuleInitFlags(startCmd)
+
+	startCmd.Flags().Bool("fork.enabled", false, "Enable forking mode")
+	startCmd.Flags().String("fork.grpc", "", "Forking remote gRPC endpoint (e.g. grpc.thor.pfc.zone:443)")
+	startCmd.Flags().String("fork.chain_id", "", "Forking remote chain-id")
+	startCmd.Flags().String("fork.chain-id", "", "Forking remote chain-id (alias)")
+	startCmd.Flags().Uint64("fork.height", 0, "Forking remote reference height (ignored; latest only)")
+	startCmd.Flags().Bool("fork.cache_enabled", true, "Enable in-memory cache for forking client")
+	startCmd.Flags().Bool("fork.cache-enabled", true, "Enable in-memory cache for forking client (alias)")
+	startCmd.Flags().Uint64("fork.cache_size", 10000, "Cache size for forking client")
+	startCmd.Flags().Uint64("fork.cache-size", 10000, "Cache size for forking client (alias)")
+	startCmd.Flags().String("fork.timeout", "60s", "Timeout for forking client requests")
+	startCmd.Flags().Uint64("fork.gas_cost_per_fetch", 0, "Gas cost per remote fetch in forking mode")
+	startCmd.Flags().Uint64("fork.gas-cost-per-fetch", 0, "Gas cost per remote fetch in forking mode (alias)")
 }
 
 func renderConfigCommand() *cobra.Command {

x/bloctopus/forkbank/keeper/iterator.go

@@ -0,0 +1 @@
+package keeper

x/bloctopus/forkbank/keeper/keeper.go

@@ -0,0 +1,130 @@
+package keeper
+
+import (
+	"context"
+
+	"cosmossdk.io/math"
+	sdk "github.com/cosmos/cosmos-sdk/types"
+	bankkeeper "github.com/cosmos/cosmos-sdk/x/bank/keeper"
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+)
+
+func NewForkingBankKeeper(base bankkeeper.BaseKeeper, cfg Config) (ForkingBankKeeper, error) {
+	client, err := NewRemoteClient(cfg.Endpoint)
+	if err != nil {
+		return ForkingBankKeeper{}, err
+	}
+	return ForkingBankKeeper{
+		BaseKeeper: base,
+		cfg:        cfg,
+		client:     client,
+	}, nil
+}
+
+func (k *ForkingBankKeeper) GetAllBalances(ctx context.Context, addr sdk.AccAddress) sdk.Coins {
+	if k.BaseKeeper.HasBalance(ctx, addr, sdk.NewCoin("rune", math.NewInt(1))) {
+		return k.BaseKeeper.GetAllBalances(ctx, addr)
+	}
+	resp, err := k.client.RemoteBalances(ctx, addr.String())
+	if err != nil || resp == nil {
+		return sdk.NewCoins()
+	}
+	return sdk.NewCoins(resp.Balances...)
+}
+
+func (k *ForkingBankKeeper) GetBalance(ctx context.Context, addr sdk.AccAddress, denom string) sdk.Coin {
+	if k.BaseKeeper.HasBalance(ctx, addr, sdk.NewCoin(denom, math.NewInt(1))) {
+		return k.BaseKeeper.GetBalance(ctx, addr, denom)
+	}
+	resp, err := k.client.RemoteBalances(ctx, addr.String())
+	if err != nil {
+		return sdk.NewCoin(denom, math.ZeroInt())
+	}
+	for _, c := range resp.Balances {
+		if c.Denom == denom {
+			return c
+		}
+	}
+	return sdk.NewCoin(denom, math.ZeroInt())
+}
+
+func (k *ForkingBankKeeper) SendCoins(ctx context.Context, fromAddr sdk.AccAddress, toAddr sdk.AccAddress, amt sdk.Coins) error {
+	return k.BaseKeeper.SendCoins(ctx, fromAddr, toAddr, amt)
+}
+
+func (k *ForkingBankKeeper) GetDenomMetaData(ctx context.Context, denom string) (banktypes.Metadata, bool) {
+	md, found := k.BaseKeeper.GetDenomMetaData(ctx, denom)
+	if found {
+		return md, true
+	}
+	k.ensureDenomMetadata(ctx)
+	return k.BaseKeeper.GetDenomMetaData(ctx, denom)
+}
+func (k *ForkingBankKeeper) ensureDenomMetadata(ctx context.Context) {
+	all := k.BaseKeeper.GetAllDenomMetaData(ctx)
+	if len(all) > 0 {
+		return
+	}
+	resp, err := k.client.RemoteDenomsMetadata(ctx)
+	if err != nil || resp == nil || len(resp.Metadatas) == 0 {
+		return
+	}
+	for _, m := range resp.Metadatas {
+		k.BaseKeeper.SetDenomMetaData(ctx, m)
+	}
+}
+func (k *ForkingBankKeeper) EnsureDenomMetadata(ctx context.Context) {
+	k.ensureDenomMetadata(ctx)
+}
+func (k *ForkingBankKeeper) DenomsMetadata(c context.Context, req *banktypes.QueryDenomsMetadataRequest) (*banktypes.QueryDenomsMetadataResponse, error) {
+	if req == nil {
+		return &banktypes.QueryDenomsMetadataResponse{Metadatas: nil}, nil
+	}
+	local := k.BaseKeeper.GetAllDenomMetaData(c)
+	if len(local) > 0 {
+		return &banktypes.QueryDenomsMetadataResponse{Metadatas: local}, nil
+	}
+	resp, err := k.client.RemoteDenomsMetadata(c)
+	if err != nil || resp == nil {
+		return &banktypes.QueryDenomsMetadataResponse{Metadatas: nil}, nil
+	}
+	return resp, nil
+}
+
+func (k *ForkingBankKeeper) DenomMetadata(c context.Context, req *banktypes.QueryDenomMetadataRequest) (*banktypes.QueryDenomMetadataResponse, error) {
+	if req == nil {
+		return &banktypes.QueryDenomMetadataResponse{}, nil
+	}
+	md, found := k.BaseKeeper.GetDenomMetaData(c, req.Denom)
+	if found {
+		return &banktypes.QueryDenomMetadataResponse{Metadata: md}, nil
+	}
+	resp, err := k.client.RemoteDenomsMetadata(c)
+	if err != nil || resp == nil {
+		return &banktypes.QueryDenomMetadataResponse{}, nil
+	}
+	for _, m := range resp.Metadatas {
+		if m.Base == req.Denom || m.Display == req.Denom || m.Name == req.Denom || m.Description == req.Denom {
+			return &banktypes.QueryDenomMetadataResponse{Metadata: m}, nil
+		}
+	}
+	return &banktypes.QueryDenomMetadataResponse{}, nil
+}
+
+
+
+
+func (k *ForkingBankKeeper) GetAllDenomMetaData(ctx context.Context) []banktypes.Metadata {
+	all := k.BaseKeeper.GetAllDenomMetaData(ctx)
+	if len(all) > 0 {
+		return all
+	}
+	resp, err := k.client.RemoteDenomsMetadata(ctx)
+	if err != nil || resp == nil || len(resp.Metadatas) == 0 {
+		return nil
+	}
+	for _, m := range resp.Metadatas {
+		k.BaseKeeper.SetDenomMetaData(ctx, m)
+	}
+	return k.BaseKeeper.GetAllDenomMetaData(ctx)
+}

x/bloctopus/forkbank/keeper/remote.go

@@ -0,0 +1,89 @@
+package keeper
+
+import (
+	"context"
+	"crypto/tls"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	"github.com/cosmos/cosmos-sdk/types/query"
+)
+
+type RemoteClient struct {
+	conn      *grpc.ClientConn
+	bankQuery banktypes.QueryClient
+	endpoint  string
+	dialed    bool
+}
+
+func NewRemoteClient(endpoint string) (*RemoteClient, error) {
+	rc := &RemoteClient{endpoint: endpoint}
+	err := rc.ensureConn()
+	if err != nil {
+		return nil, err
+	}
+	return rc, nil
+}
+
+func (c *RemoteClient) ensureConn() error {
+	if c.dialed {
+		return nil
+	}
+	creds := credentials.NewTLS(&tls.Config{
+		ServerName: "grpc.thor.pfc.zone",
+	})
+	conn, err := grpc.Dial(c.endpoint, grpc.WithTransportCredentials(creds))
+	if err != nil {
+		return err
+	}
+	c.conn = conn
+	c.bankQuery = banktypes.NewQueryClient(conn)
+	c.dialed = true
+	return nil
+}
+
+func (c *RemoteClient) Close() {
+	if c.conn != nil {
+		_ = c.conn.Close()
+	}
+}
+
+func (c *RemoteClient) RemoteBalances(ctx context.Context, addr string) (*banktypes.QueryAllBalancesResponse, error) {
+	if err := c.ensureConn(); err != nil {
+		return nil, err
+	}
+	resp, err := c.bankQuery.AllBalances(ctx, &banktypes.QueryAllBalancesRequest{Address: addr})
+	if err != nil {
+		return nil, err
+	}
+	return resp, nil
+}
+
+func (c *RemoteClient) RemoteDenomsMetadata(ctx context.Context) (*banktypes.QueryDenomsMetadataResponse, error) {
+	if err := c.ensureConn(); err != nil {
+		return nil, err
+	}
+	var all []banktypes.Metadata
+	var nextKey []byte
+	for {
+		req := &banktypes.QueryDenomsMetadataRequest{
+			Pagination: &query.PageRequest{
+				Key:        nextKey,
+				Limit:      500,
+				CountTotal: false,
+			},
+		}
+		resp, err := c.bankQuery.DenomsMetadata(ctx, req)
+		if err != nil {
+			return nil, err
+		}
+		all = append(all, resp.Metadatas...)
+		if resp.Pagination == nil || len(resp.Pagination.NextKey) == 0 {
+			break
+		}
+		nextKey = resp.Pagination.NextKey
+	}
+	return &banktypes.QueryDenomsMetadataResponse{Metadatas: all}, nil
+}

x/bloctopus/forkbank/keeper/types.go

@@ -0,0 +1,15 @@
+package keeper
+
+import (
+	bankkeeper "github.com/cosmos/cosmos-sdk/x/bank/keeper"
+)
+
+type Config struct {
+	Endpoint string
+}
+
+type ForkingBankKeeper struct {
+	bankkeeper.BaseKeeper
+	cfg    Config
+	client *RemoteClient
+}

x/bloctopus/forking/cache.go

@@ -0,0 +1 @@
+package forking

x/bloctopus/forking/client.go

@@ -0,0 +1,41 @@
+package forking
+
+import (
+	"context"
+	"crypto/tls"
+
+	banktypes "github.com/cosmos/cosmos-sdk/x/bank/types"
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/credentials"
+)
+
+type RemoteClient struct {
+	endpoint string
+	conn     *grpc.ClientConn
+	bankQ    banktypes.QueryClient
+}
+
+func NewRemoteClient(endpoint string) (*RemoteClient, error) {
+	creds := credentials.NewTLS(&tls.Config{
+		ServerName: "grpc.thor.pfc.zone",
+	})
+	conn, err := grpc.Dial(endpoint, grpc.WithTransportCredentials(creds))
+	if err != nil {
+		return nil, err
+	}
+	return &RemoteClient{
+		endpoint: endpoint,
+		conn:     conn,
+		bankQ:    banktypes.NewQueryClient(conn),
+	}, nil
+}
+
+func (c *RemoteClient) Close() {
+	if c.conn != nil {
+		_ = c.conn.Close()
+	}
+}
+
+func (c *RemoteClient) DenomsMetadata(ctx context.Context) (*banktypes.QueryDenomsMetadataResponse, error) {
+	return c.bankQ.DenomsMetadata(ctx, &banktypes.QueryDenomsMetadataRequest{})
+}

x/bloctopus/forking/flags.go

@@ -0,0 +1,13 @@
+package forking
+
+import "github.com/spf13/pflag"
+
+const (
+	flagForkingEnabled = "forking.enabled"
+	flagForkingGRPC    = "forking.grpc"
+)
+
+func AddFlags(fs *pflag.FlagSet) {
+	fs.Bool(flagForkingEnabled, true, "enable forking")
+	fs.String(flagForkingGRPC, "grpc.thor.pfc.zone:443", "forking grpc endpoint")
+}