marshaler runtime.Marshaler does not handle io.EOF when decoding

[gyuho]

Hi,

after resolving golang/protobuf/jsonpb compatibility issue, we found another issue with json decoder

Details can be found at etcd-io/etcd#5969 (comment).

To summarize

dec.Decode(&protoReq) returns io.EOF in this code (full generated code can be found https://github.com/coreos/etcd/blob/master/etcdserver/etcdserverpb/rpc.pb.gw.go#L92-L93):

func request_Watch_Watch_0(ctx context.Context, marshaler runtime.Marshaler, client WatchClient, req *http.Request, pathParams map[string]string) (Watch_WatchClient, runtime.ServerMetadata, error) {
    var metadata runtime.ServerMetadata
    stream, err := client.Watch(ctx)
    if err != nil {
        grpclog.Printf("Failed to start streaming: %v", err)
        return nil, metadata, err
    }
    dec := marshaler.NewDecoder(req.Body)
    handleSend := func() error {
        var protoReq WatchRequest
        err = dec.Decode(&protoReq)
        if err != nil {
            grpclog.Printf("Failed to decode request: %v", err)
            return err
        }
        if err = stream.Send(&protoReq); err != nil {
            grpclog.Printf("Failed to send request: %v", err)
            return err
        }
        return nil
    }

But if we manually make it handle io.EOF as below:

        dec := marshaler.NewDecoder(req.Body)
        handleSend := func() error {
                var protoReq WatchRequest
                err = dec.Decode(&protoReq)
-               if err != nil {
+               if err != nil && err != io.EOF {

where req.Body is {"create_request": {"key": "Zm9v"}}, everything works as expected.

Is there any way to make this decoder handle io.EOF?

Thanks.

[yugui]

Fixed in #191, but we'll discuss how we should handle zero-length stream.

[gyuho]

Thanks. Now our gateway server doesn't complain but now seems like it returns before interacting with gateway request?

And it works if it were written as below:

        dec := marshaler.NewDecoder(req.Body)
        handleSend := func() error {
                var protoReq WatchRequest
                err = dec.Decode(&protoReq)
-               if err != nil {
+               if err != nil && err != io.EOF {
                        grpclog.Printf("Failed to decode request: %v", err)
                        return err
                }
                if err = stream.Send(&protoReq); err != nil {
                        grpclog.Printf("Failed to send request: %v", err)
                        return err
                }

or

                var protoReq WatchRequest
                err = dec.Decode(&protoReq)
+               if err == io.EOF {
+                       return nil
+               }
                if err != nil {
                        grpclog.Printf("Failed to decode request: %v", err)
                        return err
                }
                if err = stream.Send(&protoReq); err != nil {
                        grpclog.Printf("Failed to send request: %v", err)
                        return err
                }

instead of

                var protoReq WatchRequest
                err = dec.Decode(&protoReq)
+               if err == io.EOF {
+                       return err
+               }
                if err != nil {
                        grpclog.Printf("Failed to decode request: %v", err)
                        return err
                }
                if err = stream.Send(&protoReq); err != nil {
                        grpclog.Printf("Failed to send request: %v", err)
                        return err
                }

Would it be something to be handled in zero-length stream that you mentioned?

[yugui]

@gyuho Right. The second suggestion is what I suggested in #191.
I'll take a look at this issue by myself on Monday.

[gyuho]

@yugui Another question is that why we were sending zero-length input in the first place?

        "github.com/golang/protobuf/proto"
@@ -86,7 +89,9 @@ func request_Watch_Watch_0(ctx context.Context, marshaler runtime.Marshaler, cli
                grpclog.Printf("Failed to start streaming: %v", err)
                return nil, metadata, err
        }
-       dec := marshaler.NewDecoder(req.Body)
+       bts, err := ioutil.ReadAll(req.Body)
+       fmt.Println(string(bts), err)
+       dec := marshaler.NewDecoder(bytes.NewReader(bts))

The print statement returns {"create_request": {"key": "Zm9v"}} <nil>, so the request body was not empty, but dec.Decode(&protoReq) returns io.EOF, which doesn't make sense. Is this expected? Or some kind of grpc-specific behavior?

Thanks!

[yugui]

why we were sending zero-length input in the first place

I don't think we are sending zero-length input. #200 just fixes an edge-case bug which can happen when the stream is empty.
Also the very first invocation of handleSend is different from the later ones just because let the caller know errors earlier.

The print statement returns {"create_request": {"key": "Zm9v"}} , so the request body was not empty, but dec.Decode(&protoReq) returns io.EOF, which doesn't make sense. Is this expected? Or some kind of grpc-specific behavior?

I can't reproduce. Could you give me a small reproducible case?

package main

import (
    "fmt"
    "strings"

    "github.com/grpc-ecosystem/grpc-gateway/runtime"
)

func main() {
    var m runtime.JSONPb
    dec := m.NewDecoder(strings.NewReader(`{"create_request": {"key": "Zm9v"}}`))

    msg := make(map[string]interface{})
    if err := dec.Decode(&msg); err != nil {
        fmt.Println(msg, err)
        return
    }
    fmt.Println(msg)
}

[gyuho]

Sorry for late response. Yeah I am still seeing the same behavior. I will try to have reproducible code for this issue.

Thanks.

[gyuho]

@yugui Closing this, because as you said, message itself does not send io.EOF. The problem was that we were using cURL, sending EOF when it closes the request. Thanks for help!