Golang grpc ServerStream throws error: transport: the stream is done or WriteHeader was already called

user_11156800 0
Viewed 2003

having an interesting error with my grpc server stream function and I'm tearing my hair out. Have not been able to find any possible causes from reading the grpc godoc or elsewhere online. Hoping someone more familiar with Go and grpc streams might be able to point me in the right direction.

My implementation has attempted to follow the basic Server-Streaming example on the grpc.io website.

protobuf definition in question:

service ArrayBasedCache {
  ...
  rpc GetRecord (GetRecordRequest) returns (stream MessageResponse) {}
}

message GetRecordRequest {
  string key = 1;
}

message MessageResponse {
  string message = 1;
}

grpc server handler:

func (ctlr *cacheClientController) GetRecord(req *svcgrpc.GetRecordRequest, stream svcgrpc.ArrayBasedCache_GetRecordServer) error {
    key := req.GetKey()
    if ctlr.inputChannels[key] == nil {
        return errors.New("Requested record has expired")
    }
    msgs, e1 := ctlr.client.ReadArrayRecord(key)
    if e1 != nil {
        panic(e1)
    }
    log.Printf("Messages: %v", msgs)
    for i := 0; i < len(msgs); i++ {
        log.Printf("trying to write message: %v", msgs[i])
        if e2 := stream.Send(&svcgrpc.MessageResponse{Message: msgs[i]}); e2 != nil {
            log.Printf("Writing message %d of %d to stream failed", i+1, len(msgs))
            panic(e2)
        }
    }
    return nil
}

my grpc client implementation:

func (s *GrpcService) GetRecord(key string) (svcgrpc.ArrayBasedCache_GetRecordClient, error) {
    req := &svcgrpc.GetRecordRequest{Key: key}
    ctx, cancelFunc := context.WithTimeout(context.Background(), defaultTimeout)
    defer cancelFunc()
    resp, err := s.grpcClient.GetRecord(ctx, req)
    if err != nil {
        return nil, err
    }
    return resp, nil
}

func (s *GrpcService) StreamToArray(stream svcgrpc.ArrayBasedCache_GetRecordClient) ([]string, error) {
    out := []string{}
    for {
        msg, err := stream.Recv()
        if err == io.EOF {
            break
        }
        if err != nil {
            return nil, errors.New("Unexpected error reading stream")
        }
        out = append(out, msg.Message)
    }
    return out, nil
}

The entrypoint in my test client

stream, err := c.GetRecord(testKey)
    if err != nil {
        log.Printf("Retrieving record stream %v failed", testKey)
    }
    arr, err := c.StreamToArray(stream)
    if err != nil {
        log.Printf("Failed to unmarshall message stream to array")
    }
    log.Printf("Retrieved messages %v", arr)

THE ERROR ENCOUNTERED:

2020/10/26 22:43:36 Messages: [abc def hij]
2020/10/26 22:43:36 trying to write message: abc
2020/10/26 22:43:36 Writing message 1 of 3 to stream failed
panic: rpc error: code = Internal desc = transport: transport: the stream is done or WriteHeader was already called

goroutine 82 [running]:
github.com/TasSM/app/service/controller.(*cacheClientController).GetRecord(0xc0001e4cc0, 0xc000176fc0, 0xd6ef80, 0xc00022e5f0, 0xc0001e4cc0, 0xc00039fa00)
        C:/Users/myuser/Documents/repos/app/src/service/controller/controller.go:96 +0x5ad
github.com/TasSM/app/service/svcgrpc._ArrayBasedCache_GetRecord_Handler(0xc11a80, 0xc0001e4cc0, 0xd6ca00, 0xc0001dc300, 0x11dec30, 0xc000392000)
        C:/Users/myuser/Documents/repos/app/src/service/svcgrpc/cacheservice_grpc.pb.go:192 +0x13d
google.golang.org/grpc.(*Server).processStreamingRPC(0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000, 0xc0001e4de0, 0x119fb00, 0x0, 0x0, 0x0)
        C:/Users/myuser/go/pkg/mod/google.golang.org/grpc@v1.33.0/server.go:1457 +0x15cb
google.golang.org/grpc.(*Server).handleStream(0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000, 0x0)
        C:/Users/myuser/go/pkg/mod/google.golang.org/grpc@v1.33.0/server.go:1537 +0x1309
google.golang.org/grpc.(*Server).serveStreams.func1.2(0xc000123140, 0xc0001a7a40, 0xd70300, 0xc000108c00, 0xc000392000)
        C:/Users/myuser/go/pkg/mod/google.golang.org/grpc@v1.33.0/server.go:871 +0xe0
created by google.golang.org/grpc.(*Server).serveStreams.func1
        C:/Users/myuser/go/pkg/mod/google.golang.org/grpc@v1.33.0/server.go:869 +0x349
exit status 2
grpc go stream
2 Answers

I believe the issue is within the GetRecord grpc client implementation, specifically with the context implementation you are using.

By using context.WithTimeout and by calling the defer cancelFunc() in the same method, you are basically shutting down the stream before returning from GetRecord method.

If you want to still use the context.WithTimeout implementation, don't use the cancelFunc inside the GetRecord method, but either return the cancelFunc from it, or pass the ctx to GetRecord method.

user_75675790

@eminlala identified the root cause of your stream termination.

Since a stream is meant to be long-running - returning results in realtime - in general you don't want to put a deadline on the streamed results.

If you want to put a timeout on the Dial operation that establishes the stream connection this you do via a DialOption WithTimeout:

grpc.Dial(addr, grpc.WithTimeout(defaultTimeout))

or use DialContext:

ctx, cancelFunc := context.WithTimeout(context.Background(), defaultTimeout)
defer cancelFunc()

grpc.DialContext(ctx, addr)
user_12185120
Related