feat: expose the data root inclusion proof over gRPC (#1603)

## Description

Closes #1602

#### PR checklist

- [x] Tests written/updated
- [ ] Changelog entry added in `.changelog` (we use
[unclog](https://github.com/informalsystems/unclog) to manage our
changelog)
- [ ] Updated relevant documentation (`docs/` or `spec/`) and code
comments

proto/tendermint/rpc/grpc/types.proto

@@ -6,6 +6,7 @@ import "tendermint/abci/types.proto";
 import "tendermint/types/types.proto";
 import "tendermint/p2p/types.proto";
 import "tendermint/crypto/keys.proto";
+import "tendermint/crypto/proof.proto";
 import "tendermint/types/validator.proto";
 import "google/protobuf/timestamp.proto";
 import "gogoproto/gogo.proto";
@@ -51,6 +52,15 @@ message SubscribeNewHeightsRequest {}
 
 message StatusRequest {}
 
+message DataRootInclusionProofRequest {
+  // Height the height of block we want to prove.
+  int64 height = 1;
+  // Start the start of the data commitment range containing the block.
+  uint64 start = 2;
+  // End the end exclusive of the data commitment range containing the block.
+  uint64 end = 3;
+}
+
 //----------------------------------------
 // Response types
 
@@ -128,6 +138,10 @@ message ValidatorInfo {
   int64                       voting_power = 3;
 }
 
+message DataRootInclusionProofResponse {
+  tendermint.crypto.Proof proof = 1 [(gogoproto.nullable) = false];
+}
+
 //----------------------------------------
 // Service Definition
 
@@ -144,3 +158,10 @@ service BlockAPI {
   rpc SubscribeNewHeights(SubscribeNewHeightsRequest) returns (stream NewHeightEvent);
   rpc Status(StatusRequest) returns (StatusResponse);
 }
+
+service BlobstreamAPI {
+  // DataRootInclusionProof creates an inclusion proof for the data root of block
+  // height `height` in the set of blocks defined by `start` and `end`. The range
+  // is end exclusive.
+  rpc DataRootInclusionProof(DataRootInclusionProofRequest) returns (DataRootInclusionProofResponse);
+}

rpc/core/blocks.go

@@ -235,6 +235,17 @@ func DataRootInclusionProof(
 	end uint64,
 ) (*ctypes.ResultDataRootInclusionProof, error) {
 	//nolint:gosec
+	proof, err := GenerateDataRootInclusionProof(height, start, end)
+	if err != nil {
+		return nil, err
+	}
+	return &ctypes.ResultDataRootInclusionProof{Proof: *proof}, nil
+}
+
+func GenerateDataRootInclusionProof(height int64, start, end uint64) (*merkle.Proof, error) {
+	if globalEnv == nil {
+		return nil, errors.New("global env is nil. this can only be called inside celestia-core")
+	}
 	err := validateDataRootInclusionProofRequest(uint64(height), start, end)
 	if err != nil {
 		return nil, err
@@ -247,7 +258,7 @@ func DataRootInclusionProof(
 	if err != nil {
 		return nil, err
 	}
-	return &ctypes.ResultDataRootInclusionProof{Proof: *proof}, nil
+	return proof, nil
 }
 
 // padBytes Pad bytes to given length

rpc/grpc/api.go

@@ -416,3 +416,20 @@ func (blockAPI *BlockAPI) SubscribeNewHeights(_ *SubscribeNewHeightsRequest, str
 		}
 	}
 }
+
+type BlobstreamAPI struct{}
+
+func NewBlobstreamAPI() *BlobstreamAPI {
+	return &BlobstreamAPI{}
+}
+
+func (blockAPI *BlobstreamAPI) DataRootInclusionProof(_ context.Context, req *DataRootInclusionProofRequest) (*DataRootInclusionProofResponse, error) {
+	proof, err := core.GenerateDataRootInclusionProof(req.Height, req.Start, req.End)
+	if err != nil {
+		return nil, err
+	}
+
+	return &DataRootInclusionProofResponse{
+		Proof: *proof.ToProto(),
+	}, nil
+}

rpc/grpc/client_server.go

@@ -23,13 +23,15 @@ type Config struct {
 func StartGRPCServer(ln net.Listener) error {
 	grpcServer := grpc.NewServer()
 	RegisterBroadcastAPIServer(grpcServer, &broadcastAPI{})
-	api := NewBlockAPI()
-	RegisterBlockAPIServer(grpcServer, api)
+	blockAPI := NewBlockAPI()
+	RegisterBlockAPIServer(grpcServer, blockAPI)
+	blobstreamAPI := NewBlobstreamAPI()
+	RegisterBlobstreamAPIServer(grpcServer, blobstreamAPI)
 	errCh := make(chan error, 2)
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 	go func() {
-		errCh <- api.StartNewBlockEventListener(ctx)
+		errCh <- blockAPI.StartNewBlockEventListener(ctx)
 	}()
 	go func() {
 		errCh <- grpcServer.Serve(ln)
@@ -40,7 +42,7 @@ func StartGRPCServer(ln net.Listener) error {
 		if err != nil {
 			core.GetEnvironment().Logger.Error("error stopping block api", "err", err)
 		}
-	}(api, ctx)
+	}(blockAPI, ctx)
 	// blocks until one errors or returns nil
 	return <-errCh
 }
@@ -75,3 +77,20 @@ func StartBlockAPIGRPCClient(protoAddr string, opts ...grpc.DialOption) (BlockAP
 func dialerFunc(ctx context.Context, addr string) (net.Conn, error) {
 	return cmtnet.Connect(addr)
 }
+
+// StartBlobstreamAPIGRPCClient dials the gRPC server using protoAddr and returns a new
+// BlobstreamAPIClient.
+func StartBlobstreamAPIGRPCClient(protoAddr string, opts ...grpc.DialOption) (BlobstreamAPIClient, 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 NewBlobstreamAPIClient(conn), nil
+}

rpc/grpc/grpc_test.go

@@ -392,6 +392,26 @@ func TestBlockQuery_Streaming(t *testing.T) {
 	assert.NotEqual(t, part2.BlockPart.Proof, crypto.Proof{})
 }
 
+func TestBlobstreamAPI(t *testing.T) {
+	client, err := rpctest.GetBlobstreamAPIClient()
+	require.NoError(t, err)
+	waitForHeight(t, 10)
+
+	resp, err := client.DataRootInclusionProof(
+		context.Background(),
+		&core_grpc.DataRootInclusionProofRequest{
+			Height: 6,
+			Start:  1,
+			End:    10,
+		},
+	)
+	assert.NoError(t, err)
+	assert.NotNil(t, resp)
+	assert.Equal(t, int64(9), resp.Proof.Total)
+	assert.Equal(t, int64(5), resp.Proof.Index)
+	assert.Equal(t, 4, len(resp.Proof.Aunts))
+}
+
 func waitForHeight(t *testing.T, height int64) {
 	rpcAddr := rpctest.GetConfig().RPC.ListenAddress
 	c, err := rpchttp.New(rpcAddr, "/websocket")