Repository: yarpc/yab
Pull request #328 (2021-03-09 19:45:58)

Bugfix: gRPC bidirectional stream stuck on invalid input
Bidirectional stream handler has two go-routines, one for sending requests and another one for receiving the response from the server. In case any error occurs in the `send` go-routine, the context is canceled to stop the receiving routine as both send & receive routines share the same context. But turns out context is not respected by [yarpc-go/SendMessage](https://github.com/yarpc/yarpc-go/blob/b7ba06c0dfa8658b705622e338b0ad35d14bfb78/transport/grpc/stream.go#L130) and [yarpc-go/ReceiveMessage](https://github.com/yarpc/yarpc-go/blob/b7ba06c0dfa8658b705622e338b0ad35d14bfb78/transport/grpc/stream.go#L147-L154) methods from YARPC (gRPC stream methods also do not respect context in sendMsg and recvMsg API). Instead the send and receive gRPC calls can be unblocked by canceling the context originally passed while initiating stream requests.

Fix: when an unexpected error occurs in send or receive handling, cancel the original context used during stream initiation. This closes the underlying http/2 stream, causing send and receive routines blocked on stream message APIs to unblock.

codecov[bot]: # [Codecov](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=h1) Report
> Merging [#328](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=desc) (93a58d8) into [dev](https://codecov.io/gh/yarpc/yab/commit/e80962df3b25580799432a2fbba0953de26c1ca8?el=desc) (e80962d) will **decrease** coverage by `0.00%`.
> The diff coverage is `100.00%`.

[![Impacted file tree graph](https://codecov.io/gh/yarpc/yab/pull/328/graphs/tree.svg?width=650&height=150&src=pr&token=yEgrqUTCuh)](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##              dev     #328      +/-   ##
==========================================
- Coverage   90.31%   90.31%   -0.01%     
==========================================
  Files          54       54              
  Lines        2644     2642       -2     
==========================================
- Hits         2388     2386       -2     
  Misses        180      180              
  Partials       76       76              
```


| [Impacted Files](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [handler.go](https://codecov.io/gh/yarpc/yab/pull/328/diff?src=pr&el=tree#diff-aGFuZGxlci5nbw==) | `79.62% <100.00%> (-0.25%)` | :arrow_down: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=footer). Last update [e80962d...93a58d8](https://codecov.io/gh/yarpc/yab/pull/328?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).


Patch:
diff --git a/Makefile b/Makefile
index a646e85..f9705bd 100644
--- a/Makefile
+++ b/Makefile
@@ -12,7 +12,7 @@ install:
 
 .PHONY: test
 test:
-	go test -cover -race ./...
+	go test -cover -timeout 30s -race ./...
 
 
 .PHONY: docs
@@ -28,6 +28,6 @@ docs:
 
 .PHONY: test_ci
 test_ci: install build
-	go test -race -coverprofile=cover_temp.out -coverpkg=./... ./...
+	go test -timeout 30s -race -coverprofile=cover_temp.out -coverpkg=./... ./...
 	grep -v "^github.com/yarpc/yab/testdata/*" cover_temp.out > cover.out
 	go tool cover -html=cover.out -o cover.html
diff --git a/handler.go b/handler.go
index 5b285d4..6888023 100644
--- a/handler.go
+++ b/handler.go
@@ -151,7 +151,7 @@ func makeStreamRequest(t transport.Transport, streamReq *transport.StreamRequest
 
 	switch serializer.MethodType() {
 	case encoding.BidirectionalStream:
-		return makeBidiStream(ctx, stream, streamIO, opts)
+		return makeBidiStream(ctx, cancel, stream, streamIO, opts)
 	case encoding.ClientStream:
 		return makeClientStream(ctx, stream, streamIO, opts)
 	default:
@@ -226,12 +226,9 @@ func makeClientStream(ctx context.Context, stream *yarpctransport.ClientStream,
 }
 
 // makeBidiStream starts bi-directional streaming rpc
-func makeBidiStream(ctx context.Context, stream *yarpctransport.ClientStream, streamIO StreamIO, opts StreamRequestOptions) error {
+func makeBidiStream(ctx context.Context, cancel context.CancelFunc, stream *yarpctransport.ClientStream, streamIO StreamIO, opts StreamRequestOptions) error {
 	var wg sync.WaitGroup
 	var sendErr error
-
-	ctx, cancel := context.WithCancel(ctx)
-	defer cancel()
 	reqWaiter := newIntervalWaiter(opts.Interval.Duration())
 
 	wg.Add(1)


Test patch:
diff --git a/integration_test.go b/integration_test.go
index f4793bf..3c85067 100644
--- a/integration_test.go
+++ b/integration_test.go
@@ -964,6 +964,24 @@ test: 1
 			},
 			wantErr: "Failed while reading stream input: could not parse given request body as message of type \"Foo\": Message type Foo has no known field named test_err\n",
 		},
+		{
+			desc: "bidirectional streaming with input and long timeout",
+			opts: Options{
+				ROpts: RequestOptions{
+					FileDescriptorSet: []string{"testdata/protobuf/simple/simple.proto.bin"},
+					Procedure:         "Bar/BidiStream",
+					// Ensure request timeout is longer than go test timeout (30s)
+					// See MakeFile(test_ci,test) for go test timeout.
+					Timeout:     timeMillisFlag(time.Second * 31),
+					RequestJSON: `{"test_err": 1}`,
+				},
+				TOpts: TransportOptions{
+					ServiceName: "foo",
+				},
+			},
+			expectedInput: []simple.Foo{{Test: 1}},
+			wantErr:       "Failed while reading stream input: could not parse given request body as message of type \"Foo\": Message type Foo has no known field named test_err\n",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.desc, func(t *testing.T) {