feat: support gRPC endpoints in core (#1513)

This is an implementation of a streaming API for blocks in core. 

Helps close celestiaorg/celestia-app#3421 but
not sure it entirely closes it.

It can easily be used:

```go
package main

import (
	"context"
	"fmt"
	coregrpc "github.com/tendermint/tendermint/rpc/grpc"
)

func main() {
	client := coregrpc.StartBlockAPIGRPCClient("tcp://localhost:9090")

	blockStreamer, err := client.BlockByHeight(context.Background(), &coregrpc.BlockByHeightRequest{Height: 2})
	if err != nil {
		panic(err)
	}
	blockMeta, err := client.BlockMetaByHeight(context.Background(), &coregrpc.BlockMetaByHeightRequest{Height: 2})
	if err != nil {
		panic(err)
	}
	parts := make([]*core.Part, 0)
	for i := 0; i < int(blockMeta.BlockMeta.BlockID.PartSetHeader.Total); i++ {
		resp, err := blockStreamer.Recv()
		if err != nil {
			panic(err)
		}
		parts = append(parts, resp.BlockPart)
		if resp.IsLast && i < int(blockMeta.BlockMeta.BlockID.PartSetHeader.Total)-1 {
			panic("couldn't get all parts")
		} else if resp.IsLast {
			break
		}
	}

	h := types.NewPartSetFromHeader(types.PartSetHeader{
		Total: blockMeta.BlockMeta.BlockID.PartSetHeader.Total,
		Hash:  blockMeta.BlockMeta.BlockID.PartSetHeader.Hash,
	})

	for _, part := range parts {
		ok, err := h.AddPart(&types.Part{
			Index: part.Index,
			Bytes: part.Bytes,
			Proof: merkle.Proof{
				Total:    part.Proof.Total,
				Index:    part.Proof.Index,
				LeafHash: part.Proof.LeafHash,
				Aunts:    part.Proof.Aunts,
			},
		})
		if err != nil {
			panic(err)
		}
		if !ok {
			panic("not okey")
		}
	}
	pbb := new(core.Block)
	bz, err := io.ReadAll(h.GetReader())
	if err != nil {
		panic(err)
	}
	err = proto.Unmarshal(bz, pbb)
	if err != nil {
		panic(err)
	}
	block, err := types.BlockFromProto(pbb)
	if err != nil {
		panic(err)
	}
	fmt.Println(block)

	// get a commit
	commit, err := client.Commit(context.Background(), &coregrpc.CommitRequest{Height: 10})
	if err != nil {
		panic(err)
	}
	fmt.Println(commit)

	// listen for new heights
	streamer, err := client.SubscribeNewHeights(context.Background(), &coregrpc.SubscribeNewHeightsRequest{})
	if err != nil {
		panic(err)
	}
	for {
		resp, err := streamer.Recv()
		if err != nil {
			panic(err)
		}
		fmt.Println(resp)
	}
}
```

Ps: I didn't add the tests because I didn't find a direct way of mocking
the environment without polluting the rest of the repo (exporting some
methods, adding new helpers, etc). And I think since the implementation
is simple, just querying the block/state stores for results, it's fine
to leave it untested.

---------

Co-authored-by: Rootul P <[email protected]>

Author: rach-id

buf.yaml

@@ -9,6 +9,8 @@ lint:
     - BASIC
     - FILE_LOWER_SNAKE_CASE
     - UNARY_RPC
+  except:
+    - RPC_NO_SERVER_STREAMING
   ignore:
     - gogoproto
 breaking:

proto/buf.yaml

@@ -10,3 +10,5 @@ lint:
     - BASIC
     - FILE_LOWER_SNAKE_CASE
     - UNARY_RPC
+  except:
+    - RPC_NO_SERVER_STREAMING

proto/tendermint/rpc/grpc/types.pb.go

@@ -3,6 +3,12 @@ package tendermint.rpc.grpc;
 option  go_package = "github.com/tendermint/tendermint/rpc/grpc;coregrpc";
 
 import "tendermint/abci/types.proto";
+import "tendermint/types/types.proto";
+import "tendermint/p2p/types.proto";
+import "tendermint/crypto/keys.proto";
+import "tendermint/types/validator.proto";
+import "google/protobuf/timestamp.proto";
+import "gogoproto/gogo.proto";
 
 //----------------------------------------
 // Request types
@@ -13,6 +19,38 @@ message RequestBroadcastTx {
   bytes tx = 1;
 }
 
+message BlockByHashRequest {
+  bytes hash  = 1;
+  bool  prove = 2;
+}
+
+message BlockByHeightRequest {
+  // Height the requested block height.
+  // If height is equal to 0, the latest height stored in the block store
+  // will be used.
+  int64 height = 1;
+  // Prove set to true to return the parts proofs.
+  bool prove = 2;
+}
+
+message CommitRequest {
+  // Height the requested block commit height.
+  // If height is equal to 0, the latest height stored in the block store
+  // will be used.
+  int64 height = 1;
+}
+
+message ValidatorSetRequest {
+  // Height the requested validator set height.
+  // If height is equal to 0, the latest height stored in the block store
+  // will be used.
+  int64 height = 1;
+}
+
+message SubscribeNewHeightsRequest {}
+
+message StatusRequest {}
+
 //----------------------------------------
 // Response types
 
@@ -23,10 +61,86 @@ message ResponseBroadcastTx {
   tendermint.abci.ResponseDeliverTx deliver_tx = 2;
 }
 
+message StreamedBlockByHashResponse {
+  tendermint.types.Part         block_part    = 1;
+  // Commit is only set in the first part, and
+  // it stays nil in the remaining ones.
+  tendermint.types.Commit       commit        = 2;
+  // ValidatorSet is only set in the first part, and
+  // it stays nil in the remaining ones.
+  tendermint.types.ValidatorSet validator_set = 3;
+  bool                          is_last       = 4;
+}
+
+message StreamedBlockByHeightResponse {
+  tendermint.types.Part block_part = 1;
+  // Commit is only set in the first part, and
+  // it stays nil in the remaining ones.
+  tendermint.types.Commit commit = 2;
+  // ValidatorSet is only set in the first part, and
+  // it stays nil in the remaining ones.
+  tendermint.types.ValidatorSet validator_set = 3;
+  bool                       is_last    = 4;
+}
+
+message CommitResponse {
+  tendermint.types.Commit commit = 1;
+}
+
+message ValidatorSetResponse {
+  // ValidatorSet the requested validator set.
+  tendermint.types.ValidatorSet validator_set = 1;
+  // Height the height corresponding to the returned
+  // validator set.
+  int64 height = 2;
+}
+
+message NewHeightEvent {
+  int64 height = 1;
+  bytes hash   = 2;
+}
+
+message StatusResponse {
+  tendermint.p2p.DefaultNodeInfo node_info      = 1;
+  SyncInfo                       sync_info      = 2;
+  ValidatorInfo                  validator_info = 3;
+}
+
+message SyncInfo {
+  bytes                     latest_block_hash   = 1;
+  bytes                     latest_app_hash     = 2;
+  int64                     latest_block_height = 3;
+  google.protobuf.Timestamp latest_block_time   = 4
+      [(gogoproto.nullable) = false, (gogoproto.stdtime) = true];
+
+  bytes                     earliest_block_hash   = 5;
+  bytes                     earliest_app_hash     = 6;
+  int64                     earliest_block_height = 7;
+  google.protobuf.Timestamp earliest_block_time   = 8
+      [(gogoproto.nullable) = false, (gogoproto.stdtime) = true];
+
+  bool catching_up = 9;
+}
+
+message ValidatorInfo {
+  bytes                       address      = 1;
+  tendermint.crypto.PublicKey pub_key      = 2;
+  int64                       voting_power = 3;
+}
+
 //----------------------------------------
 // Service Definition
 
 service BroadcastAPI {
   rpc Ping(RequestPing) returns (ResponsePing);
   rpc BroadcastTx(RequestBroadcastTx) returns (ResponseBroadcastTx);
 }
+
+service BlockAPI {
+  rpc BlockByHash(BlockByHashRequest) returns (stream StreamedBlockByHashResponse);
+  rpc BlockByHeight(BlockByHeightRequest) returns (stream StreamedBlockByHeightResponse);
+  rpc Commit(CommitRequest) returns (CommitResponse);
+  rpc ValidatorSet(ValidatorSetRequest) returns (ValidatorSetResponse);
+  rpc SubscribeNewHeights(SubscribeNewHeightsRequest) returns (stream NewHeightEvent);
+  rpc Status(StatusRequest) returns (StatusResponse);
+}

proto/tendermint/rpc/grpc/types.proto

@@ -13,7 +13,7 @@ import (
 // Status returns CometBFT status including node info, pubkey, latest block
 // hash, app hash, block height and time.
 // More: https://docs.cometbft.com/v0.34/rpc/#/Info/status
-func Status(ctx *rpctypes.Context) (*ctypes.ResultStatus, error) {
+func Status(_ *rpctypes.Context) (*ctypes.ResultStatus, error) {
 	var (
 		earliestBlockHeight   int64
 		earliestBlockHash     cmtbytes.HexBytes

rpc/core/status.go

@@ -2,10 +2,24 @@ package coregrpc
 
 import (
 	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/tendermint/tendermint/libs/pubsub"
+
+	"github.com/tendermint/tendermint/crypto/encoding"
+
+	"github.com/tendermint/tendermint/proto/tendermint/crypto"
+
+	"github.com/tendermint/tendermint/libs/rand"
 
 	abci "github.com/tendermint/tendermint/abci/types"
-	core "github.com/tendermint/tendermint/rpc/core"
+	"github.com/tendermint/tendermint/proto/tendermint/types"
+	"github.com/tendermint/tendermint/rpc/core"
 	rpctypes "github.com/tendermint/tendermint/rpc/jsonrpc/types"
+	eventstypes "github.com/tendermint/tendermint/types"
 )
 
 type broadcastAPI struct {
@@ -37,3 +51,372 @@ func (bapi *broadcastAPI) BroadcastTx(ctx context.Context, req *RequestBroadcast
 		},
 	}, nil
 }
+
+type BlockAPI struct {
+	sync.Mutex
+	heightListeners      map[chan NewHeightEvent]struct{}
+	newBlockSubscription eventstypes.Subscription
+	subscriptionID       string
+	subscriptionQuery    pubsub.Query
+}
+
+func NewBlockAPI() *BlockAPI {
+	return &BlockAPI{
+		heightListeners:   make(map[chan NewHeightEvent]struct{}, 1000),
+		subscriptionID:    fmt.Sprintf("block-api-subscription-%s", rand.Str(6)),
+		subscriptionQuery: eventstypes.EventQueryNewBlock,
+	}
+}
+
+func (blockAPI *BlockAPI) StartNewBlockEventListener(ctx context.Context) error {
+	env := core.GetEnvironment()
+	if blockAPI.newBlockSubscription == nil {
+		var err error
+		blockAPI.newBlockSubscription, err = env.EventBus.Subscribe(
+			ctx,
+			blockAPI.subscriptionID,
+			blockAPI.subscriptionQuery,
+			500,
+		)
+		if err != nil {
+			env.Logger.Error("Failed to subscribe to new blocks", "err", err)
+			return err
+		}
+	}
+	for {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-blockAPI.newBlockSubscription.Cancelled():
+			env.Logger.Error("cancelled grpc subscription. retrying")
+			ok, err := blockAPI.retryNewBlocksSubscription(ctx)
+			if err != nil {
+				return err
+			}
+			if !ok {
+				// this will happen when the context is done. we can stop here
+				return nil
+			}
+		case event, ok := <-blockAPI.newBlockSubscription.Out():
+			if !ok {
+				env.Logger.Error("new blocks subscription closed. re-subscribing")
+				ok, err := blockAPI.retryNewBlocksSubscription(ctx)
+				if err != nil {
+					return err
+				}
+				if !ok {
+					// this will happen when the context is done. we can stop here
+					return nil
+				}
+				continue
+			}
+			newBlockEvent, ok := event.Events()[eventstypes.EventTypeKey]
+			if !ok || len(newBlockEvent) == 0 || newBlockEvent[0] != eventstypes.EventNewBlock {
+				continue
+			}
+			data, ok := event.Data().(eventstypes.EventDataNewBlock)
+			if !ok {
+				env.Logger.Error("couldn't cast event data to new block")
+				return fmt.Errorf("couldn't cast event data to new block. Events: %s", event.Events())
+			}
+			blockAPI.broadcastToListeners(ctx, data.Block.Height, data.Block.Hash())
+		}
+	}
+}
+
+// RetryAttempts the number of retry times when the subscription is closed.
+const RetryAttempts = 6
+
+// SubscriptionCapacity the maximum number of pending blocks in the subscription.
+const SubscriptionCapacity = 500
+
+func (blockAPI *BlockAPI) retryNewBlocksSubscription(ctx context.Context) (bool, error) {
+	env := core.GetEnvironment()
+	ticker := time.NewTicker(time.Second)
+	defer ticker.Stop()
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+	for i := 1; i < RetryAttempts; i++ {
+		select {
+		case <-ctx.Done():
+			return false, nil
+		case <-ticker.C:
+			var err error
+			blockAPI.newBlockSubscription, err = env.EventBus.Subscribe(
+				ctx,
+				fmt.Sprintf("block-api-subscription-%s", rand.Str(6)),
+				blockAPI.subscriptionQuery,
+				SubscriptionCapacity,
+			)
+			if err != nil {
+				env.Logger.Error("Failed to subscribe to new blocks. retrying", "err", err, "retry_number", i)
+			} else {
+				return true, nil
+			}
+		}
+	}
+	return false, errors.New("couldn't recover from failed blocks subscription. stopping listeners")
+}
+
+func (blockAPI *BlockAPI) broadcastToListeners(ctx context.Context, height int64, hash []byte) {
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+	for ch := range blockAPI.heightListeners {
+		func() {
+			defer func() {
+				if r := recover(); r != nil {
+					// logging the error then removing the heights listener
+					core.GetEnvironment().Logger.Debug("failed to write to heights listener", "err", r)
+					blockAPI.removeHeightListener(ch)
+				}
+			}()
+			select {
+			case <-ctx.Done():
+				return
+			case ch <- NewHeightEvent{Height: height, Hash: hash}:
+			}
+		}()
+	}
+}
+
+func (blockAPI *BlockAPI) addHeightListener() chan NewHeightEvent {
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+	ch := make(chan NewHeightEvent, 50)
+	blockAPI.heightListeners[ch] = struct{}{}
+	return ch
+}
+
+func (blockAPI *BlockAPI) removeHeightListener(ch chan NewHeightEvent) {
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+	delete(blockAPI.heightListeners, ch)
+}
+
+func (blockAPI *BlockAPI) closeAllListeners() {
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+	if blockAPI.heightListeners == nil {
+		// if this is nil, then there is no need to close anything
+		return
+	}
+	for channel := range blockAPI.heightListeners {
+		delete(blockAPI.heightListeners, channel)
+	}
+}
+
+// Stop cleans up the BlockAPI instance by closing all listeners
+// and ensuring no further events are processed.
+func (blockAPI *BlockAPI) Stop(ctx context.Context) error {
+	blockAPI.Lock()
+	defer blockAPI.Unlock()
+
+	// close all height listeners
+	blockAPI.closeAllListeners()
+
+	var err error
+	// stop the events subscription
+	if blockAPI.newBlockSubscription != nil {
+		err = core.GetEnvironment().EventBus.Unsubscribe(ctx, blockAPI.subscriptionID, blockAPI.subscriptionQuery)
+		blockAPI.newBlockSubscription = nil
+	}
+
+	core.GetEnvironment().Logger.Info("gRPC streaming API has been stopped")
+	return err
+}
+
+func (blockAPI *BlockAPI) BlockByHash(req *BlockByHashRequest, stream BlockAPI_BlockByHashServer) error {
+	blockStore := core.GetEnvironment().BlockStore
+	blockMeta := blockStore.LoadBlockMetaByHash(req.Hash)
+	if blockMeta == nil {
+		return fmt.Errorf("nil block meta for block hash %d", req.Hash)
+	}
+	commit := blockStore.LoadBlockCommit(blockMeta.Header.Height)
+	if commit == nil {
+		return fmt.Errorf("nil commit for block hash %d", req.Hash)
+	}
+	protoCommit := commit.ToProto()
+
+	validatorSet, err := core.GetEnvironment().StateStore.LoadValidators(blockMeta.Header.Height)
+	if err != nil {
+		return err
+	}
+	protoValidatorSet, err := validatorSet.ToProto()
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < int(blockMeta.BlockID.PartSetHeader.Total); i++ {
+		part, err := blockStore.LoadBlockPart(blockMeta.Header.Height, i).ToProto()
+		if err != nil {
+			return err
+		}
+		if part == nil {
+			return fmt.Errorf("nil block part %d for block hash %d", i, req.Hash)
+		}
+		if !req.Prove {
+			part.Proof = crypto.Proof{}
+		}
+		isLastPart := i == int(blockMeta.BlockID.PartSetHeader.Total)-1
+		resp := StreamedBlockByHashResponse{
+			BlockPart: part,
+			IsLast:    isLastPart,
+		}
+		if i == 0 {
+			resp.ValidatorSet = protoValidatorSet
+			resp.Commit = protoCommit
+		}
+		err = stream.Send(&resp)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (blockAPI *BlockAPI) BlockByHeight(req *BlockByHeightRequest, stream BlockAPI_BlockByHeightServer) error {
+	blockStore := core.GetEnvironment().BlockStore
+	height := req.Height
+	if height == 0 {
+		height = blockStore.Height()
+	}
+
+	blockMeta := blockStore.LoadBlockMeta(height)
+	if blockMeta == nil {
+		return fmt.Errorf("nil block meta for height %d", height)
+	}
+
+	commit := blockStore.LoadSeenCommit(height)
+	if commit == nil {
+		return fmt.Errorf("nil block commit for height %d", height)
+	}
+	protoCommit := commit.ToProto()
+
+	validatorSet, err := core.GetEnvironment().StateStore.LoadValidators(height)
+	if err != nil {
+		return err
+	}
+	protoValidatorSet, err := validatorSet.ToProto()
+	if err != nil {
+		return err
+	}
+
+	for i := 0; i < int(blockMeta.BlockID.PartSetHeader.Total); i++ {
+		part, err := blockStore.LoadBlockPart(height, i).ToProto()
+		if err != nil {
+			return err
+		}
+		if part == nil {
+			return fmt.Errorf("nil block part %d for height %d", i, height)
+		}
+		if !req.Prove {
+			part.Proof = crypto.Proof{}
+		}
+		isLastPart := i == int(blockMeta.BlockID.PartSetHeader.Total)-1
+		resp := StreamedBlockByHeightResponse{
+			BlockPart: part,
+			IsLast:    isLastPart,
+		}
+		if i == 0 {
+			resp.ValidatorSet = protoValidatorSet
+			resp.Commit = protoCommit
+		}
+		err = stream.Send(&resp)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (blockAPI *BlockAPI) Status(_ context.Context, _ *StatusRequest) (*StatusResponse, error) {
+	status, err := core.Status(nil)
+	if err != nil {
+		return nil, err
+	}
+
+	protoPubKey, err := encoding.PubKeyToProto(status.ValidatorInfo.PubKey)
+	if err != nil {
+		return nil, err
+	}
+	return &StatusResponse{
+		NodeInfo: status.NodeInfo.ToProto(),
+		SyncInfo: &SyncInfo{
+			LatestBlockHash:     status.SyncInfo.LatestBlockHash,
+			LatestAppHash:       status.SyncInfo.LatestAppHash,
+			LatestBlockHeight:   status.SyncInfo.LatestBlockHeight,
+			LatestBlockTime:     status.SyncInfo.LatestBlockTime,
+			EarliestBlockHash:   status.SyncInfo.EarliestBlockHash,
+			EarliestAppHash:     status.SyncInfo.EarliestAppHash,
+			EarliestBlockHeight: status.SyncInfo.EarliestBlockHeight,
+			EarliestBlockTime:   status.SyncInfo.EarliestBlockTime,
+			CatchingUp:          status.SyncInfo.CatchingUp,
+		},
+		ValidatorInfo: &ValidatorInfo{
+			Address:     status.ValidatorInfo.Address,
+			PubKey:      &protoPubKey,
+			VotingPower: status.ValidatorInfo.VotingPower,
+		},
+	}, nil
+}
+
+func (blockAPI *BlockAPI) Commit(_ context.Context, req *CommitRequest) (*CommitResponse, error) {
+	blockStore := core.GetEnvironment().BlockStore
+	height := req.Height
+	if height == 0 {
+		height = blockStore.Height()
+	}
+	commit := blockStore.LoadSeenCommit(height)
+	if commit == nil {
+		return nil, fmt.Errorf("nil block commit for height %d", height)
+	}
+	protoCommit := commit.ToProto()
+
+	return &CommitResponse{
+		Commit: &types.Commit{
+			Height:     protoCommit.Height,
+			Round:      protoCommit.Round,
+			BlockID:    protoCommit.BlockID,
+			Signatures: protoCommit.Signatures,
+		},
+	}, nil
+}
+
+func (blockAPI *BlockAPI) ValidatorSet(_ context.Context, req *ValidatorSetRequest) (*ValidatorSetResponse, error) {
+	blockStore := core.GetEnvironment().BlockStore
+	height := req.Height
+	if height == 0 {
+		height = blockStore.Height()
+	}
+	validatorSet, err := core.GetEnvironment().StateStore.LoadValidators(height)
+	if err != nil {
+		return nil, err
+	}
+	protoValidatorSet, err := validatorSet.ToProto()
+	if err != nil {
+		return nil, err
+	}
+	return &ValidatorSetResponse{
+		ValidatorSet: protoValidatorSet,
+		Height:       height,
+	}, nil
+}
+
+func (blockAPI *BlockAPI) SubscribeNewHeights(_ *SubscribeNewHeightsRequest, stream BlockAPI_SubscribeNewHeightsServer) error {
+	heightListener := blockAPI.addHeightListener()
+	defer blockAPI.removeHeightListener(heightListener)
+
+	for {
+		select {
+		case event, ok := <-heightListener:
+			if !ok {
+				return errors.New("blocks subscription closed from the service side")
+			}
+			if err := stream.Send(&event); err != nil {
+				return err
+			}
+		case <-stream.Context().Done():
+			return nil
+		}
+	}
+}

rpc/grpc/api.go

@@ -3,6 +3,8 @@ package coregrpc
 import (
 	"net"
 
+	"github.com/tendermint/tendermint/rpc/core"
+
 	"golang.org/x/net/context"
 	"google.golang.org/grpc"
 	"google.golang.org/grpc/credentials/insecure"
@@ -21,7 +23,26 @@ type Config struct {
 func StartGRPCServer(ln net.Listener) error {
 	grpcServer := grpc.NewServer()
 	RegisterBroadcastAPIServer(grpcServer, &broadcastAPI{})
-	return grpcServer.Serve(ln)
+	api := NewBlockAPI()
+	RegisterBlockAPIServer(grpcServer, api)
+	errCh := make(chan error, 2)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	go func() {
+		errCh <- api.StartNewBlockEventListener(ctx)
+	}()
+	go func() {
+		errCh <- grpcServer.Serve(ln)
+	}()
+	defer grpcServer.GracefulStop()
+	defer func(api *BlockAPI, ctx context.Context) {
+		err := api.Stop(ctx)
+		if err != nil {
+			core.GetEnvironment().Logger.Error("error stopping block api", "err", err)
+		}
+	}(api, ctx)
+	// blocks until one errors or returns nil
+	return <-errCh
 }
 
 // StartGRPCClient dials the gRPC server using protoAddr and returns a new
@@ -34,6 +55,23 @@ func StartGRPCClient(protoAddr string) BroadcastAPIClient {
 	return NewBroadcastAPIClient(conn)
 }
 
+// StartBlockAPIGRPCClient dials the gRPC server using protoAddr and returns a new
+// BlockAPIClient.
+func StartBlockAPIGRPCClient(protoAddr string, opts ...grpc.DialOption) (BlockAPIClient, error) {
+	if len(opts) == 0 {
+		opts = append(opts, grpc.WithTransportCredentials(insecure.NewCredentials()))
+	}
+	opts = append(opts, grpc.WithContextDialer(dialerFunc))
+	conn, err := grpc.Dial( //nolint:staticcheck
+		protoAddr,
+		opts...,
+	)
+	if err != nil {
+		return nil, err
+	}
+	return NewBlockAPIClient(conn), nil
+}
+
 func dialerFunc(ctx context.Context, addr string) (net.Conn, error) {
 	return cmtnet.Connect(addr)
 }

rpc/grpc/client_server.go

@@ -1,9 +1,19 @@
+//nolint:dupl
 package coregrpc_test
 
 import (
 	"context"
 	"os"
 	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/tendermint/tendermint/libs/rand"
+	"github.com/tendermint/tendermint/proto/tendermint/crypto"
+	rpcclient "github.com/tendermint/tendermint/rpc/client"
+	rpchttp "github.com/tendermint/tendermint/rpc/client/http"
+	"github.com/tendermint/tendermint/rpc/core"
+	"github.com/tendermint/tendermint/types"
 
 	"github.com/stretchr/testify/require"
 
@@ -33,3 +43,359 @@ func TestBroadcastTx(t *testing.T) {
 	require.EqualValues(t, 0, res.CheckTx.Code)
 	require.EqualValues(t, 0, res.DeliverTx.Code)
 }
+
+func setupClient(t *testing.T) core_grpc.BlockAPIClient {
+	client, err := rpctest.GetBlockAPIClient()
+	require.NoError(t, err)
+	return client
+}
+
+func TestBlockByHash(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 2)
+	expectedBlockMeta := core.GetEnvironment().BlockStore.LoadBlockMeta(1)
+	require.NotNil(t, expectedBlockMeta)
+
+	// query the block along with the part proofs
+	res, err := client.BlockByHash(context.Background(), &core_grpc.BlockByHashRequest{
+		Hash:  expectedBlockMeta.BlockID.Hash,
+		Prove: true,
+	})
+	require.NoError(t, err)
+
+	part, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part.BlockPart)
+	require.NotNil(t, part.ValidatorSet)
+	require.NotNil(t, part.Commit)
+
+	assert.NotEqual(t, part.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part.Commit.Height, expectedBlockMeta.Header.Height)
+
+	// query the block along without the part proofs
+	res, err = client.BlockByHash(context.Background(), &core_grpc.BlockByHashRequest{
+		Hash:  expectedBlockMeta.BlockID.Hash,
+		Prove: false,
+	})
+	require.NoError(t, err)
+
+	part, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part.BlockPart)
+	require.NotNil(t, part.ValidatorSet)
+	require.NotNil(t, part.Commit)
+
+	assert.Equal(t, part.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part.Commit.Height, expectedBlockMeta.Header.Height)
+}
+
+func TestCommit(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 2)
+	expectedBlockCommit := core.GetEnvironment().BlockStore.LoadSeenCommit(1)
+
+	res, err := client.Commit(context.Background(), &core_grpc.CommitRequest{
+		Height: 1,
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, expectedBlockCommit.BlockID.Hash.Bytes(), res.Commit.BlockID.Hash)
+}
+
+func TestLatestCommit(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 3)
+
+	res, err := client.Commit(context.Background(), &core_grpc.CommitRequest{
+		Height: 0,
+	})
+	require.NoError(t, err)
+
+	assert.Greater(t, res.Commit.Height, int64(2))
+}
+
+func TestValidatorSet(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 2)
+	expectedValidatorSet, err := core.GetEnvironment().StateStore.LoadValidators(1)
+	require.NoError(t, err)
+
+	res, err := client.ValidatorSet(context.Background(), &core_grpc.ValidatorSetRequest{
+		Height: 1,
+	})
+	require.NoError(t, err)
+
+	assert.Equal(t, len(expectedValidatorSet.Validators), len(res.ValidatorSet.Validators))
+}
+
+func TestLatestValidatorSet(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 3)
+
+	res, err := client.ValidatorSet(context.Background(), &core_grpc.ValidatorSetRequest{
+		Height: 0,
+	})
+	require.NoError(t, err)
+
+	assert.Greater(t, res.Height, int64(2))
+}
+
+func TestStatus(t *testing.T) {
+	client := setupClient(t)
+	expectedStatus, err := core.Status(nil)
+	require.NoError(t, err)
+
+	res, err := client.Status(context.Background(), &core_grpc.StatusRequest{})
+	require.NoError(t, err)
+	assert.Equal(t, string(expectedStatus.NodeInfo.DefaultNodeID), res.NodeInfo.DefaultNodeID)
+}
+
+func TestSubscribeNewHeights(t *testing.T) {
+	client := setupClient(t)
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	stream, err := client.SubscribeNewHeights(ctx, &core_grpc.SubscribeNewHeightsRequest{})
+	require.NoError(t, err)
+	store := core.GetEnvironment().BlockStore
+
+	go func() {
+		listenedHeightsCount := 0
+		defer func() {
+			assert.Greater(t, listenedHeightsCount, 0)
+		}()
+		for {
+			res, err := stream.Recv()
+			if ctx.Err() != nil {
+				return
+			}
+			require.NoError(t, err)
+			require.Greater(t, res.Height, int64(0))
+			assert.Equal(t, store.LoadBlockMeta(res.Height).BlockID.Hash.Bytes(), res.Hash)
+			listenedHeightsCount++
+		}
+	}()
+
+	time.Sleep(5 * time.Second)
+}
+
+func TestBlockByHash_Streaming(t *testing.T) {
+	client := setupClient(t)
+
+	// send a big transaction that would result in a block
+	// containing multiple block parts
+	txRes, err := rpctest.GetGRPCClient().BroadcastTx(
+		context.Background(),
+		&core_grpc.RequestBroadcastTx{Tx: rand.NewRand().Bytes(1000000)},
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, 0, txRes.CheckTx.Code)
+	require.EqualValues(t, 0, txRes.DeliverTx.Code)
+
+	var expectedBlockMeta types.BlockMeta
+	for i := int64(1); i < 500; i++ {
+		waitForHeight(t, i+1)
+		blockMeta := core.GetEnvironment().BlockStore.LoadBlockMeta(i)
+		if blockMeta.BlockID.PartSetHeader.Total > 1 {
+			expectedBlockMeta = *blockMeta
+			break
+		}
+	}
+
+	// query the block without the part proofs
+	res, err := client.BlockByHash(context.Background(), &core_grpc.BlockByHashRequest{
+		Hash:  expectedBlockMeta.BlockID.Hash,
+		Prove: false,
+	})
+	require.NoError(t, err)
+
+	part1, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part1.BlockPart)
+	require.NotNil(t, part1.ValidatorSet)
+	require.NotNil(t, part1.Commit)
+
+	assert.Equal(t, part1.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part1.Commit.Height, expectedBlockMeta.Header.Height)
+
+	part2, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part2.BlockPart)
+	require.Nil(t, part2.ValidatorSet)
+	require.Nil(t, part2.Commit)
+
+	assert.Equal(t, part2.BlockPart.Proof, crypto.Proof{})
+
+	// query the block along with the part proofs
+	res, err = client.BlockByHash(context.Background(), &core_grpc.BlockByHashRequest{
+		Hash:  expectedBlockMeta.BlockID.Hash,
+		Prove: true,
+	})
+	require.NoError(t, err)
+
+	part1, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part1.BlockPart)
+	require.NotNil(t, part1.ValidatorSet)
+	require.NotNil(t, part1.Commit)
+
+	assert.NotEqual(t, part1.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part1.Commit.Height, expectedBlockMeta.Header.Height)
+
+	part2, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part2.BlockPart)
+	require.Nil(t, part2.ValidatorSet)
+	require.Nil(t, part2.Commit)
+
+	assert.NotEqual(t, part2.BlockPart.Proof, crypto.Proof{})
+}
+
+func TestBlockByHeight(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 2)
+	expectedBlockMeta := core.GetEnvironment().BlockStore.LoadBlockMeta(1)
+
+	// query the block along with the part proofs
+	res, err := client.BlockByHeight(context.Background(), &core_grpc.BlockByHeightRequest{
+		Height: expectedBlockMeta.Header.Height,
+		Prove:  true,
+	})
+	require.NoError(t, err)
+
+	part, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part.BlockPart)
+	require.NotNil(t, part.ValidatorSet)
+	require.NotNil(t, part.Commit)
+
+	assert.NotEqual(t, part.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part.Commit.Height, expectedBlockMeta.Header.Height)
+
+	// query the block along without the part proofs
+	res, err = client.BlockByHeight(context.Background(), &core_grpc.BlockByHeightRequest{
+		Height: expectedBlockMeta.Header.Height,
+		Prove:  false,
+	})
+	require.NoError(t, err)
+
+	part, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part.BlockPart)
+	require.NotNil(t, part.ValidatorSet)
+	require.NotNil(t, part.Commit)
+
+	assert.Equal(t, part.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part.Commit.Height, expectedBlockMeta.Header.Height)
+}
+
+func TestLatestBlockByHeight(t *testing.T) {
+	client := setupClient(t)
+	waitForHeight(t, 2)
+
+	// query the block along with the part proofs
+	res, err := client.BlockByHeight(context.Background(), &core_grpc.BlockByHeightRequest{
+		Height: 0,
+	})
+	require.NoError(t, err)
+
+	part, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part.BlockPart)
+	require.NotNil(t, part.ValidatorSet)
+	require.NotNil(t, part.Commit)
+
+	assert.Greater(t, part.Commit.Height, int64(2))
+}
+
+func TestBlockQuery_Streaming(t *testing.T) {
+	client := setupClient(t)
+
+	// send a big transaction that would result in a block
+	// containing multiple block parts
+	txRes, err := rpctest.GetGRPCClient().BroadcastTx(
+		context.Background(),
+		&core_grpc.RequestBroadcastTx{Tx: rand.NewRand().Bytes(1000000)},
+	)
+	require.NoError(t, err)
+	require.EqualValues(t, 0, txRes.CheckTx.Code)
+	require.EqualValues(t, 0, txRes.DeliverTx.Code)
+
+	var expectedBlockMeta types.BlockMeta
+	for i := int64(1); i < 500; i++ {
+		waitForHeight(t, i+1)
+		blockMeta := core.GetEnvironment().BlockStore.LoadBlockMeta(i)
+		if blockMeta.BlockID.PartSetHeader.Total > 1 {
+			expectedBlockMeta = *blockMeta
+			break
+		}
+	}
+
+	// query the block without the part proofs
+	res, err := client.BlockByHeight(context.Background(), &core_grpc.BlockByHeightRequest{
+		Height: expectedBlockMeta.Header.Height,
+		Prove:  false,
+	})
+	require.NoError(t, err)
+
+	part1, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part1.BlockPart)
+	require.NotNil(t, part1.ValidatorSet)
+	require.NotNil(t, part1.Commit)
+
+	assert.Equal(t, part1.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part1.Commit.Height, expectedBlockMeta.Header.Height)
+
+	part2, err := res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part2.BlockPart)
+	require.Nil(t, part2.ValidatorSet)
+	require.Nil(t, part2.Commit)
+
+	assert.Equal(t, part2.BlockPart.Proof, crypto.Proof{})
+
+	// query the block along with the part proofs
+	res, err = client.BlockByHeight(context.Background(), &core_grpc.BlockByHeightRequest{
+		Height: expectedBlockMeta.Header.Height,
+		Prove:  true,
+	})
+	require.NoError(t, err)
+
+	part1, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part1.BlockPart)
+	require.NotNil(t, part1.ValidatorSet)
+	require.NotNil(t, part1.Commit)
+
+	assert.NotEqual(t, part1.BlockPart.Proof, crypto.Proof{})
+	assert.Equal(t, part1.Commit.Height, expectedBlockMeta.Header.Height)
+
+	part2, err = res.Recv()
+	require.NoError(t, err)
+
+	require.NotNil(t, part2.BlockPart)
+	require.Nil(t, part2.ValidatorSet)
+	require.Nil(t, part2.Commit)
+
+	assert.NotEqual(t, part2.BlockPart.Proof, crypto.Proof{})
+}
+
+func waitForHeight(t *testing.T, height int64) {
+	rpcAddr := rpctest.GetConfig().RPC.ListenAddress
+	c, err := rpchttp.New(rpcAddr, "/websocket")
+	require.NoError(t, err)
+	err = rpcclient.WaitForHeight(c, height, nil)
+	require.NoError(t, err)
+}

rpc/grpc/grpc_test.go

@@ -117,6 +117,15 @@ func GetGRPCClient() core_grpc.BroadcastAPIClient {
 	return core_grpc.StartGRPCClient(grpcAddr)
 }
 
+func GetBlockAPIClient() (core_grpc.BlockAPIClient, error) {
+	grpcAddr := globalConfig.RPC.GRPCListenAddress
+	client, err := core_grpc.StartBlockAPIGRPCClient(grpcAddr)
+	if err != nil {
+		return nil, err
+	}
+	return client, nil
+}
+
 // StartTendermint starts a test CometBFT server in a go routine and returns when it is initialized
 func StartTendermint(app abci.Application, opts ...func(*Options)) *nm.Node {
 	nodeOpts := defaultOptions

rpc/grpc/types.pb.go

@@ -271,6 +271,27 @@ func (ps *PartSet) Total() uint32 {
 }
 
 func (ps *PartSet) AddPart(part *Part) (bool, error) {
+	if part == nil {
+		return false, fmt.Errorf("nil part")
+	}
+
+	// The proof should be compatible with the number of parts.
+	if part.Proof.Total != int64(ps.total) {
+		return false, ErrPartSetInvalidProof
+	}
+
+	// Check hash proof
+	if part.Proof.Verify(ps.Hash(), part.Bytes) != nil {
+		return false, ErrPartSetInvalidProof
+	}
+
+	return ps.AddPartWithoutProof(part)
+}
+
+func (ps *PartSet) AddPartWithoutProof(part *Part) (bool, error) {
+	if part == nil {
+		return false, fmt.Errorf("nil part")
+	}
 	if ps == nil {
 		return false, nil
 	}
@@ -287,16 +308,6 @@ func (ps *PartSet) AddPart(part *Part) (bool, error) {
 		return false, nil
 	}
 
-	// The proof should be compatible with the number of parts.
-	if part.Proof.Total != int64(ps.total) {
-		return false, ErrPartSetInvalidProof
-	}
-
-	// Check hash proof
-	if part.Proof.Verify(ps.Hash(), part.Bytes) != nil {
-		return false, ErrPartSetInvalidProof
-	}
-
 	// Add part
 	ps.parts[part.Index] = part
 	ps.partsBitArray.SetIndex(int(part.Index), true)