refactor(state): Remove need for core RPC conn and rely only on gRPC …

…for state reads (#3359)

Resolves #3357 

Blocked on patch release from app including
celestiaorg/celestia-app#3424

Corresponding docs PR celestiaorg/docs#1558

go.mod

@@ -15,7 +15,6 @@ require (
 	github.com/celestiaorg/nmt v0.20.0
 	github.com/celestiaorg/rsmt2d v0.13.1
 	github.com/cosmos/cosmos-sdk v0.46.16
-	github.com/cosmos/cosmos-sdk/api v0.1.0
 	github.com/cristalhq/jwt v1.2.0
 	github.com/dgraph-io/badger/v4 v4.2.1-0.20240106094458-1c417aa3799c
 	github.com/etclabscore/go-openrpc-reflect v0.0.37

go.sum

@@ -481,8 +481,6 @@ github.com/cosmos/btcutil v1.0.5 h1:t+ZFcX77LpKtDBhjucvnOH8C2l2ioGsBNEQ3jef8xFk=
 github.com/cosmos/btcutil v1.0.5/go.mod h1:IyB7iuqZMJlthe2tkIFL33xPyzbFYP0XVdS8P5lUPis=
 github.com/cosmos/cosmos-proto v1.0.0-alpha8 h1:d3pCRuMYYvGA5bM0ZbbjKn+AoQD4A7dyNG2wzwWalUw=
 github.com/cosmos/cosmos-proto v1.0.0-alpha8/go.mod h1:6/p+Bc4O8JKeZqe0VqUGTX31eoYqemTT4C1hLCWsO7I=
-github.com/cosmos/cosmos-sdk/api v0.1.0 h1:xfSKM0e9p+EJTMQnf5PbWE6VT8ruxTABIJ64Rd064dE=
-github.com/cosmos/cosmos-sdk/api v0.1.0/go.mod h1:CupqQBskAOiTXO1XDZ/wrtWzN/wTxUvbQmOqdUhR8wI=
 github.com/cosmos/go-bip39 v0.0.0-20180819234021-555e2067c45d/go.mod h1:tSxLoYXyBmiFeKpvmq4dzayMdCjCnu8uqmCysIGBT2Y=
 github.com/cosmos/go-bip39 v1.0.0 h1:pcomnQdrdH22njcAatO0yWojsUnCO3y2tNoV1cb6hHY=
 github.com/cosmos/go-bip39 v1.0.0/go.mod h1:RNJv0H/pOIVgxw6KS7QeX2a0Uo0aKUlfhZ4xuwvCdJw=

nodebuilder/state/core.go

@@ -28,8 +28,7 @@ func coreAccessor(
 	*modfraud.ServiceBreaker[*state.CoreAccessor, *header.ExtendedHeader],
 	error,
 ) {
-	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, corecfg.IP, corecfg.RPCPort, corecfg.GRPCPort,
-		opts...)
+	ca, err := state.NewCoreAccessor(keyring, string(keyname), sync, corecfg.IP, corecfg.GRPCPort, opts...)
 
 	sBreaker := &modfraud.ServiceBreaker[*state.CoreAccessor, *header.ExtendedHeader]{
 		Service:   ca,

state/core_access.go

@@ -9,8 +9,8 @@ import (
 	"time"
 
 	sdkErrors "cosmossdk.io/errors"
-	"github.com/cosmos/cosmos-sdk/api/tendermint/abci"
 	nodeservice "github.com/cosmos/cosmos-sdk/client/grpc/node"
+	"github.com/cosmos/cosmos-sdk/client/grpc/tmservice"
 	"github.com/cosmos/cosmos-sdk/crypto/keyring"
 	storetypes "github.com/cosmos/cosmos-sdk/store/types"
 	sdktypes "github.com/cosmos/cosmos-sdk/types"
@@ -22,8 +22,7 @@ import (
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 	logging "github.com/ipfs/go-log/v2"
 	"github.com/tendermint/tendermint/crypto/merkle"
-	rpcclient "github.com/tendermint/tendermint/rpc/client"
-	"github.com/tendermint/tendermint/rpc/client/http"
+	"github.com/tendermint/tendermint/proto/tendermint/crypto"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
 
@@ -78,15 +77,14 @@ type CoreAccessor struct {
 
 	getter libhead.Head[*header.ExtendedHeader]
 
-	stakingCli  stakingtypes.QueryClient
-	feeGrantCli feegrant.QueryClient
-	rpcCli      rpcclient.ABCIClient
+	stakingCli   stakingtypes.QueryClient
+	feeGrantCli  feegrant.QueryClient
+	abciQueryCli tmservice.ServiceClient
 
 	prt *merkle.ProofRuntime
 
 	coreConn *grpc.ClientConn
 	coreIP   string
-	rpcPort  string
 	grpcPort string
 
 	// these fields are mutatable and thus need to be protected by a mutex
@@ -113,7 +111,6 @@ func NewCoreAccessor(
 	keyname string,
 	getter libhead.Head[*header.ExtendedHeader],
 	coreIP,
-	rpcPort string,
 	grpcPort string,
 	options ...Option,
 ) (*CoreAccessor, error) {
@@ -136,7 +133,6 @@ func NewCoreAccessor(
 		addr:     addr,
 		getter:   getter,
 		coreIP:   coreIP,
-		rpcPort:  rpcPort,
 		grpcPort: grpcPort,
 		prt:      prt,
 	}
@@ -166,16 +162,13 @@ func (ca *CoreAccessor) Start(ctx context.Context) error {
 	ca.coreConn = client
 
 	// create the staking query client
-	stakingCli := stakingtypes.NewQueryClient(ca.coreConn)
-	ca.stakingCli = stakingCli
+	ca.stakingCli = stakingtypes.NewQueryClient(ca.coreConn)
 	ca.feeGrantCli = feegrant.NewQueryClient(ca.coreConn)
+
 	// create ABCI query client
-	cli, err := http.New(fmt.Sprintf("http://%s:%s", ca.coreIP, ca.rpcPort), "/websocket")
-	if err != nil {
-		return err
-	}
-	ca.rpcCli = cli
+	ca.abciQueryCli = tmservice.NewServiceClient(ca.coreConn)
 
+	// set up signer to handle tx submission
 	ca.signer, err = ca.setupSigner(ctx)
 	if err != nil {
 		log.Warnw("failed to set up signer, check if node's account is funded", "err", err)
@@ -334,26 +327,22 @@ func (ca *CoreAccessor) BalanceForAddress(ctx context.Context, addr Address) (*B
 	// TODO @renaynay: once https://github.com/cosmos/cosmos-sdk/pull/12674 is merged, use this method
 	// instead
 	prefixedAccountKey := append(banktypes.CreateAccountBalancesPrefix(addr.Bytes()), []byte(app.BondDenom)...)
-	abciReq := abci.RequestQuery{
+	req := &tmservice.ABCIQueryRequest{
+		Data: prefixedAccountKey,
 		// TODO @renayay: once https://github.com/cosmos/cosmos-sdk/pull/12674 is merged, use const instead
 		Path:   fmt.Sprintf("store/%s/key", banktypes.StoreKey),
 		Height: int64(head.Height() - 1),
-		Data:   prefixedAccountKey,
 		Prove:  true,
 	}
-	opts := rpcclient.ABCIQueryOptions{
-		Height: abciReq.GetHeight(),
-		Prove:  abciReq.GetProve(),
-	}
-	result, err := ca.rpcCli.ABCIQueryWithOptions(ctx, abciReq.GetPath(), abciReq.GetData(), opts)
-	if err != nil {
+
+	result, err := ca.abciQueryCli.ABCIQuery(ctx, req)
+	if err != nil || result.GetCode() != 0 {
+		err = fmt.Errorf("failed to query for balance: %w; result log: %s", err, result.GetLog())
 		return nil, err
 	}
-	if !result.Response.IsOK() {
-		return nil, sdkErrorToGRPCError(result.Response)
-	}
+
 	// unmarshal balance information
-	value := result.Response.Value
+	value := result.GetValue()
 	// if the value returned is empty, the account balance does not yet exist
 	if len(value) == 0 {
 		log.Errorf("balance for account %s does not exist at block height %d", addr.String(), head.Height()-1)
@@ -366,9 +355,25 @@ func (ca *CoreAccessor) BalanceForAddress(ctx context.Context, addr Address) (*B
 	if !ok {
 		return nil, fmt.Errorf("cannot convert %s into sdktypes.Int", string(value))
 	}
+
+	if result.GetProofOps() == nil {
+		return nil, fmt.Errorf("failed to get proofs for balance of address %s", addr.String())
+	}
+
 	// verify balance
+	proofOps := &crypto.ProofOps{
+		Ops: make([]crypto.ProofOp, len(result.ProofOps.Ops)),
+	}
+	for i, proofOp := range result.ProofOps.Ops {
+		proofOps.Ops[i] = crypto.ProofOp{
+			Type: proofOp.Type,
+			Key:  proofOp.Key,
+			Data: proofOp.Data,
+		}
+	}
+
 	err = ca.prt.VerifyValueFromKeys(
-		result.Response.GetProofOps(),
+		proofOps,
 		head.AppHash,
 		[][]byte{
 			[]byte(banktypes.StoreKey),

state/core_access_test.go

@@ -32,12 +32,11 @@ func TestSubmitPayForBlob(t *testing.T) {
 	appConf.MinGasPrices = fmt.Sprintf("0.002%s", app.BondDenom)
 
 	config := testnode.DefaultConfig().WithTendermintConfig(tmCfg).WithAppConfig(appConf).WithAccounts(accounts)
-	cctx, rpcAddr, grpcAddr := testnode.NewNetwork(t, config)
+	cctx, _, grpcAddr := testnode.NewNetwork(t, config)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	ca, err := NewCoreAccessor(cctx.Keyring, accounts[0], nil, "127.0.0.1", extractPort(rpcAddr),
-		extractPort(grpcAddr))
+	ca, err := NewCoreAccessor(cctx.Keyring, accounts[0], nil, "127.0.0.1", extractPort(grpcAddr))
 	require.NoError(t, err)
 	// start the accessor
 	err = ca.Start(ctx)

state/integration_test.go

@@ -7,6 +7,7 @@ import (
 	"testing"
 
 	"github.com/cosmos/cosmos-sdk/client/flags"
+	"github.com/cosmos/cosmos-sdk/client/grpc/tmservice"
 	sdk "github.com/cosmos/cosmos-sdk/types"
 	stakingtypes "github.com/cosmos/cosmos-sdk/x/staking/types"
 	"github.com/stretchr/testify/require"
@@ -48,23 +49,23 @@ func (s *IntegrationTestSuite) SetupSuite() {
 	s.cctx = core.StartTestNodeWithConfig(s.T(), cfg)
 	s.accounts = cfg.Accounts
 
-	accessor, err := NewCoreAccessor(s.cctx.Keyring, s.accounts[0], localHeader{s.cctx.Client}, "", "", "")
+	accessor, err := NewCoreAccessor(s.cctx.Keyring, s.accounts[0], localHeader{s.cctx.Client}, "", "")
 	require.NoError(s.T(), err)
-	setClients(accessor, s.cctx.GRPCClient, s.cctx.Client)
+	setClients(accessor, s.cctx.GRPCClient)
 	s.accessor = accessor
 
 	// required to ensure the Head request is non-nil
 	_, err = s.cctx.WaitForHeight(3)
 	require.NoError(s.T(), err)
 }
 
-func setClients(ca *CoreAccessor, conn *grpc.ClientConn, abciCli rpcclient.ABCIClient) {
+func setClients(ca *CoreAccessor, conn *grpc.ClientConn) {
 	ca.coreConn = conn
 	// create the staking query client
 	stakingCli := stakingtypes.NewQueryClient(ca.coreConn)
 	ca.stakingCli = stakingCli
 
-	ca.rpcCli = abciCli
+	ca.abciQueryCli = tmservice.NewServiceClient(ca.coreConn)
 }
 
 func (s *IntegrationTestSuite) TearDownSuite() {