Connect unary should return unavailable instead of unimplemented for …

…`io.EOF` errors from the transport (#776)

This adds a test that can reproduce the issue where unary RPCs would
return an `unimplemented` code when an `unavailable` code was more
appropriate. Basically, it was mistakenly interpreting network EOF
errors as "the server did not send us any response messages which is
a cardinality violation" instead of "the server connection broke so the
server response is unavailable".

There are still cases where an `io.EOF` could likely be misinterpreted
as a cardinality issue: if it the connection is severed _after_ the
headers have been received but _before_ the first byte of the first
envelope has been received. But the main case where this issue has been
reported was for unary RPCs in the Connect protocol, which does not
suffer from that issue, since it does not use envelopes.

.golangci.yml

@@ -19,8 +19,6 @@ linters-settings:
  importas:
  no-unaliased: true
  alias:
- - pkg: connectrpc.com/connect
- alias: connect
  - pkg: connectrpc.com/connect/internal/gen/connect/ping/v1
  alias: pingv1
  varnamelen:

client_ext_test.go

@@ -31,7 +31,7 @@ import (
  "testing"
  "time"
 
- connect "connectrpc.com/connect"
+ "connectrpc.com/connect"
  "connectrpc.com/connect/internal/assert"
  pingv1 "connectrpc.com/connect/internal/gen/connect/ping/v1"
  "connectrpc.com/connect/internal/gen/connect/ping/v1/pingv1connect"
@@ -227,6 +227,58 @@ func TestGetNoContentHeaders(t *testing.T) {
  assert.Equal(t, http.MethodGet, unaryReq.HTTPMethod())
 }
 
+func TestConnectionDropped(t *testing.T) {
+ t.Parallel()
+ ctx := context.Background()
+ for _, protocol := range []string{connect.ProtocolConnect, connect.ProtocolGRPC, connect.ProtocolGRPCWeb} {
+ var opts []connect.ClientOption
+ switch protocol {
+ case connect.ProtocolGRPC:
+ opts = []connect.ClientOption{connect.WithGRPC()}
+ case connect.ProtocolGRPCWeb:
+ opts = []connect.ClientOption{connect.WithGRPCWeb()}
+ }
+ t.Run(protocol, func(t *testing.T) {
+ t.Parallel()
+ httpClient := httpClientFunc(func(_ *http.Request) (*http.Response, error) {
+ return nil, io.EOF
+ })
+ client := pingv1connect.NewPingServiceClient(
+ httpClient,
+ "http://1.2.3.4",
+ opts...,
+ )
+ t.Run("unary", func(t *testing.T) {
+ t.Parallel()
+ req := connect.NewRequest[pingv1.PingRequest](nil)
+ _, err := client.Ping(ctx, req)
+ assert.NotNil(t, err)
+ if !assert.Equal(t, connect.CodeOf(err), connect.CodeUnavailable) {
+ t.Logf("err = %v\n%#v", err, err)
+ }
+ })
+ t.Run("stream", func(t *testing.T) {
+ t.Parallel()
+ req := connect.NewRequest[pingv1.CountUpRequest](nil)
+ svrStream, err := client.CountUp(ctx, req)
+ if err == nil {
+ t.Cleanup(func() {
+ assert.Nil(t, svrStream.Close())
+ })
+ if !assert.False(t, svrStream.Receive()) {
+ return
+ }
+ err = svrStream.Err()
+ }
+ assert.NotNil(t, err)
+ if !assert.Equal(t, connect.CodeOf(err), connect.CodeUnavailable) {
+ t.Logf("err = %v\n%#v", err, err)
+ }
+ })
+ })
+ }
+}
+
 func TestSpecSchema(t *testing.T) {
  t.Parallel()
  mux := http.NewServeMux()
@@ -762,3 +814,9 @@ func addUnrecognizedBytes[M proto.Message](msg M, data []byte) M {
  msg.ProtoReflect().SetUnknown(data)
  return msg
 }
+
+type httpClientFunc func(*http.Request) (*http.Response, error)
+
+func (fn httpClientFunc) Do(req *http.Request) (*http.Response, error) {
+ return fn(req)
+}

duplex_http_call.go

@@ -308,6 +308,11 @@ func (d *duplexHTTPCall) makeRequest() {
  // pipe. Write's check for io.ErrClosedPipe and will convert this to io.EOF.
  response, err := d.httpClient.Do(d.request) //nolint:bodyclose
  if err != nil {
+ if errors.Is(err, io.EOF) {
+ // We use io.EOF as a sentinel in many places and don't want this
+ // transport error to be confused for those other situations.
+ err = io.ErrUnexpectedEOF
+ }
  err = wrapIfContextError(err)
  err = wrapIfLikelyH2CNotConfiguredError(d.request, err)
  err = wrapIfLikelyWithGRPCNotUsedError(err)