feat: support gRPC endpoints in core

[rach-id]

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:

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.

[rach-id]

Just found out that once we enable the app grpc, these endpoints get overridden, I'll be looking into it

[cmwaters]

Thanks for E2E testing this change @rach-id. Have a few questions before we merge

[cmwaters]

Is it possible to prefix these types as StreamedBlockByHashResponse for example so it's more obvious that this type is streamed. Also could we add a comment that the commit, validator set and block meta is only populated for the first response in that stream? (I guess this is how it works right).

Also why BlockMeta and not just Header?

[rach-id]

Is it possible to prefix these types as StreamedBlockByHashResponse for example so it's more obvious that this type is streamed. Also could we add a comment that the commit, validator set and block meta is only populated for the first response in that stream? (I guess this is how it works right).

ff4f343

Also why BlockMeta and not just Header?

that's what node team asked for

[rach-id]

removed block meta here: 4ca8293

[rootulp]

LGTM. Left a few non-blocking questions / suggestions

[rootulp]

[question] should this log an error and terminate instead of continuing?

Suggested change

	env.Logger.Debug("couldn't cast event data to new block")
	continue
	env.Logger.Error("couldn't cast event data to new block")
	return

[rach-id]

I didn't want to have the gRPC logs polluting the output but it's a good point. I will handle this in a separate PR where I will be doing more testing and adding more logic for failures etc. I didn't want to include everything in the same PR to reduce complexity

[cmwaters]

But this case should never happen right? That you subscribe for an event of a certain type and you get an unexpected type back. Shouldn't this panic as it's an internal developer error

[rach-id]

Panicking would mean stopping the node. is this something we want to do?

That's a separate issue I want to handle in a subsequent PR. If the gRPC connection panics, the node goes down. So I am thinking of having recovers around here or add them to the cosmos-sdk.

[rootulp]

[question] Should we extract a constant for RETRY_ATTEMPTS = 5

Suggested change

	for i := 1; i < 6; i++ {
	for i := 0; i < RETRY_ATTEMPTS; i++ {

[rach-id]

yes, we could. I can extract it if you want

[rootulp]

IMO an extracted constant is easier to read but this comment is an optional nit and not blocking.

[rach-id]

a8f8a21

[cmwaters]

What does this get used for?

[rach-id]

for the new blocks subscription. whenever there is a new block, this message is broadcasted to listeners

[cmwaters]

Sorry, I meant specifically the hash

[rach-id]

aaah, it was also a requirement from node as they preferred to have the hash too. It's not currently used in the implementation, but maybe they have some ideas around that to reduce the number of requests

[cmwaters]

But this case should never happen right? That you subscribe for an event of a certain type and you get an unexpected type back. Shouldn't this panic as it's an internal developer error

[cmwaters]

I would prefer this function returns an error that can be logged out in grpccore.StartGRPCServer(...). I don't think we should allow this service to be able to fail silently. You can still run this in a go routine but you should pass the error back in a channel and return the error if either this or the Serve method fails.

[rach-id]

the issue with this is that the cosmos-sdk doesn't listen for errors after the server starts listening (for about 5sec or so). Making it listen would require more changes, which I tried to reduce given we want to reduce diff with upstream.

Current way it will log errors but not stop the node or anything.

[cmwaters]

having this error in the method is cleaner than adding logs for every return. There are also a few returns that don't error.

It also looks like this is called in cometbft not cosmos-sdk (I see it in node.go). Am I missing something?

[rach-id]

having this error in the method is cleaner than adding logs for every return. There are also a few returns that don't error.

So you mean we return an error from this method and also from the others and refactor the comsos-sdk to act when an error happens?

It also looks like this is called in cometbft not cosmos-sdk (I see it in node.go). Am I missing something?

this is called in cometBFT when you run plain comet (for example in e2e). But when you run an application, the cosmos-sdk overrides that server and calls this in there.

[cmwaters]

It looks like so long as the RPC addr and GRPC addr in the config.toml is populated it will run it in CometBFT as well.

Ideally all go routines shut down gracefully when the node is shut down or errors. I would want the gRPC which is now something that the DA network will depend on to fail noisily. This could be with the consensus node also spinning down or something else that notifies the operator that there is a problem

[rach-id]

stopping the node is on the table? I guess it's good to stop the node if bridge nodes are connected to full nodes. But if they're connected to validators, it shouldn't kill the node as it will stop a validator

[cmwaters]

What causes a block subscription to be cancelled?

[rach-id]

when it's manually cancelled and maybe other cases I am not aware of

[cmwaters]

Could we have context with a cancel and cancel this go routine whenever grpcServer.Serve() returns (as this implies that something has called Stop or GracefulStop)

[rach-id]

where would that context come from? passed as a parameter?

thing is only cosmos-sdk will call this, and there is no context there to pass.

[cmwaters]

cosmos-sdk calls StartGRPCServer. I see it called also by the cometbft node OnStart. Is it being called twice?

Regardless what I was proposing is:

func RunGRPCServer(ln net.Listener) error {
	grpcServer := grpc.NewServer()
	RegisterBroadcastAPIServer(grpcServer, &broadcastAPI{})
	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()
        // blocks until one errors or returns nil
	return <-errCh
}

This means that both processes get shut down when either of them errors or terminates

[rach-id]

cosmos-sdk calls StartGRPCServer. I see it called also by the cometbft node OnStart. Is it being called twice?

when starting comet, the OnStart gets called. When using the cosmos-sdk, that server is overriden a new one is created.

This means that both processes get shut down when either of them errors or terminates

so the issue is having them shutdown both instead of one failing and the other staying up, right?

[cmwaters]

yes and just making sure all errors are collected and logged

[rach-id]

1ea65f7