This is an automated email from the ASF dual-hosted git repository.
Alanxtl pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/dubbo-go.git
The following commit(s) were added to refs/heads/develop by this push:
new 9c5f8203f fix(triple): make stream response close non-blocking (#3466)
9c5f8203f is described below
commit 9c5f8203fdce09650290f90af627da8d15420cb7
Author: Mochimia <[email protected]>
AuthorDate: Tue Jul 7 10:52:31 2026 +0800
fix(triple): make stream response close non-blocking (#3466)
* fix(triple): make stream response close non-blocking
* test(triple): stabilize stream close regression tests
* test(triple): avoid shadowing error in stream close test
* test(triple): clarify stream close lifecycle comments
---
.../triple/triple_protocol/duplex_http_call.go | 7 +-
.../triple_protocol/duplex_http_call_test.go | 90 ++++++++
.../triple_protocol/stream_close_ext_test.go | 251 +++++++++++++++++++++
3 files changed, 344 insertions(+), 4 deletions(-)
diff --git a/protocol/triple/triple_protocol/duplex_http_call.go
b/protocol/triple/triple_protocol/duplex_http_call.go
index ef4ecc393..1ef3f0372 100644
--- a/protocol/triple/triple_protocol/duplex_http_call.go
+++ b/protocol/triple/triple_protocol/duplex_http_call.go
@@ -193,10 +193,9 @@ func (d *duplexHTTPCall) CloseRead() error {
if d.response == nil {
return nil
}
- if err := discard(d.response.Body); err != nil {
- return wrapIfRSTError(err)
- }
- // Return incoming data via context, if set outgoing data.
+ // Do not read the response body here. CloseRead must return even when
the
+ // peer leaves a streaming response open; callers that need trailers
should
+ // read to EOF before closing.
if ExtractFromOutgoingContext(d.ctx) != nil {
newIncomingContext(d.ctx, d.ResponseTrailer())
}
diff --git a/protocol/triple/triple_protocol/duplex_http_call_test.go
b/protocol/triple/triple_protocol/duplex_http_call_test.go
index f48e22f0c..3da5683cc 100644
--- a/protocol/triple/triple_protocol/duplex_http_call_test.go
+++ b/protocol/triple/triple_protocol/duplex_http_call_test.go
@@ -15,18 +15,22 @@
package triple_protocol
import (
+ "context"
"io"
"net/http"
"net/http/httptest"
"net/url"
"sync"
"testing"
+ "time"
)
import (
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/assert"
)
+const closeTestTimeout = time.Second
+
// newTestDuplexClientCall builds a duplexHTTPCall for a client-streaming RPC
// backed by a server that simply drains the request body and replies 200. It
is
// the minimal setup needed to exercise the Write/CloseWrite/CloseRead
lifecycle
@@ -182,3 +186,89 @@ func TestDuplexHTTPCallConcurrentWriteCloseRace(t
*testing.T) {
_ = call.CloseRead()
}
}
+
+// TestDuplexHTTPCallCloseReadDoesNotDrainResponseBody verifies that CloseRead
+// closes the response body directly instead of reading until EOF. This guards
+// the stream-close hang fixed in connect-go#791.
+//
+// A streaming peer may keep the response open after it stops sending data, so
+// draining the body during close can block forever. Callers that need final
+// trailers should read to EOF before closing the read side.
+func TestDuplexHTTPCallCloseReadDoesNotDrainResponseBody(t *testing.T) {
+ t.Parallel()
+ body := newBlockingReadCloser()
+ call := &duplexHTTPCall{
+ ctx: context.Background(),
+ responseReady: make(chan struct{}),
+ response: &http.Response{
+ Body: body,
+ Trailer: make(http.Header),
+ },
+ }
+ close(call.responseReady)
+
+ done := make(chan error, 1)
+ go func() {
+ done <- call.CloseRead()
+ }()
+
+ select {
+ case err := <-done:
+ assert.Nil(t, err)
+ case <-body.readStarted:
+ body.unblock()
+ assert.Nil(t, <-done)
+ t.Fatal("CloseRead attempted to drain the response body")
+ case <-time.After(closeTestTimeout):
+ body.unblock()
+ assert.Nil(t, <-done)
+ t.Fatal("CloseRead blocked while closing the response body")
+ }
+
+ select {
+ case <-body.closed:
+ default:
+ t.Fatal("CloseRead did not close the response body")
+ }
+}
+
+// blockingReadCloser records attempted reads and then blocks. It lets the test
+// catch drain-before-close behavior without depending on a real stalled
stream.
+type blockingReadCloser struct {
+ readStarted chan struct{}
+ readUnblock chan struct{}
+ closed chan struct{}
+
+ startReadOnce sync.Once
+ unblockOnce sync.Once
+ closeOnce sync.Once
+}
+
+func newBlockingReadCloser() *blockingReadCloser {
+ return &blockingReadCloser{
+ readStarted: make(chan struct{}),
+ readUnblock: make(chan struct{}),
+ closed: make(chan struct{}),
+ }
+}
+
+func (b *blockingReadCloser) Read([]byte) (int, error) {
+ b.startReadOnce.Do(func() {
+ close(b.readStarted)
+ })
+ <-b.readUnblock
+ return 0, io.EOF
+}
+
+func (b *blockingReadCloser) Close() error {
+ b.closeOnce.Do(func() {
+ close(b.closed)
+ })
+ return nil
+}
+
+func (b *blockingReadCloser) unblock() {
+ b.unblockOnce.Do(func() {
+ close(b.readUnblock)
+ })
+}
diff --git a/protocol/triple/triple_protocol/stream_close_ext_test.go
b/protocol/triple/triple_protocol/stream_close_ext_test.go
new file mode 100644
index 000000000..8dc176e46
--- /dev/null
+++ b/protocol/triple/triple_protocol/stream_close_ext_test.go
@@ -0,0 +1,251 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package triple_protocol_test
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "sync"
+ "testing"
+ "time"
+)
+
+import (
+ triple "dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol"
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/assert"
+ pingv1
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1"
+
"dubbo.apache.org/dubbo-go/v3/protocol/triple/triple_protocol/internal/gen/proto/connect/ping/v1/pingv1connect"
+)
+
+const closeTestTimeout = time.Second
+
+// TestServerStreamCloseDoesNotDrainResponse verifies that Close returns
+// promptly after the client stops reading a server-streaming response.
+//
+// The server sends one message and then keeps the stream open. Close should
+// release the response body without waiting for the stream to reach EOF.
+func TestServerStreamCloseDoesNotDrainResponse(t *testing.T) {
+ release, unblock := newCloseRelease()
+ t.Cleanup(unblock)
+ sent := make(chan struct{})
+ client := newCloseLifecyclePingClient(t, &pluggablePingServer{
+ countUp: func(ctx context.Context, req *triple.Request, stream
*triple.ServerStream) error {
+ if err := stream.Send(&pingv1.CountUpResponse{Number:
1}); err != nil {
+ return err
+ }
+ close(sent)
+ <-release
+ return nil
+ },
+ })
+
+ stream, err := client.CountUp(context.Background(),
triple.NewRequest(&pingv1.CountUpRequest{}))
+ assert.Nil(t, err)
+
+ select {
+ case <-sent:
+ case <-time.After(closeTestTimeout):
+ t.Fatal("server did not send the first response before close
verification")
+ }
+
+ if !stream.Receive(&pingv1.CountUpResponse{}) {
+ t.Fatalf("failed to receive first response before close
verification: %v", stream.Err())
+ }
+ msg := stream.Msg().(*pingv1.CountUpResponse)
+ assert.Equal(t, msg.Number, int64(1))
+
+ done := make(chan error, 1)
+ go func() {
+ done <- stream.Close()
+ }()
+ assert.Nil(t, assertCloseReturnsPromptly(t, done, unblock,
"ServerStreamForClient.Close blocked while draining the response"))
+}
+
+// TestBidiStreamCloseResponseDoesNotDrainResponse verifies that CloseResponse
+// does not wait for EOF on a bidi response stream.
+//
+// After one successful receive, the server keeps the response side open.
+// Closing the client receive side should still return promptly.
+func TestBidiStreamCloseResponseDoesNotDrainResponse(t *testing.T) {
+ release, unblock := newCloseRelease()
+ t.Cleanup(unblock)
+ received := make(chan struct{})
+ sent := make(chan struct{})
+ client := newCloseLifecyclePingClient(t, &pluggablePingServer{
+ cumSum: func(ctx context.Context, stream *triple.BidiStream)
error {
+ req := &pingv1.CumSumRequest{}
+ if err := stream.Receive(req); err != nil {
+ return err
+ }
+ close(received)
+ if err := stream.Send(&pingv1.CumSumResponse{Sum:
req.Number}); err != nil {
+ return err
+ }
+ close(sent)
+ <-release
+ return nil
+ },
+ })
+
+ stream, err := client.CumSum(context.Background())
+ assert.Nil(t, err)
+ if err := stream.Send(&pingv1.CumSumRequest{Number: 2}); err != nil {
+ t.Fatalf("failed to send setup request before close
verification: %v", err)
+ }
+
+ select {
+ case <-received:
+ case <-time.After(closeTestTimeout):
+ t.Fatal("server did not receive the setup request before close
verification")
+ }
+ select {
+ case <-sent:
+ case <-time.After(closeTestTimeout):
+ t.Fatal("server did not send the setup response before close
verification")
+ }
+
+ res := &pingv1.CumSumResponse{}
+ if err := stream.Receive(res); err != nil {
+ t.Fatalf("failed to receive setup response before close
verification: %v", err)
+ }
+ assert.Equal(t, res.Sum, int64(2))
+
+ done := make(chan error, 1)
+ go func() {
+ done <- stream.CloseResponse()
+ }()
+ assert.Nil(t, assertCloseReturnsPromptly(t, done, unblock,
"BidiStreamForClient.CloseResponse blocked while draining the response"))
+ assert.Nil(t, stream.CloseRequest())
+}
+
+// TestBidiStreamCloseResponseAfterServerStopsReading covers the case where the
+// server returns an error before consuming the rest of the request stream.
+//
+// Once the client has received that error, CloseResponse should be local
cleanup
+// and should not wait for the server to continue the stream.
+func TestBidiStreamCloseResponseAfterServerStopsReading(t *testing.T) {
+ serverReceived := make(chan struct{})
+ serverReturn := make(chan struct{})
+ client := newCloseLifecyclePingClient(t, &pluggablePingServer{
+ cumSum: func(ctx context.Context, stream *triple.BidiStream)
error {
+ req := &pingv1.CumSumRequest{}
+ if err := stream.Receive(req); err != nil {
+ return err
+ }
+ close(serverReceived)
+ <-serverReturn
+ return triple.NewError(triple.CodeUnavailable,
errors.New("server stopped reading"))
+ },
+ })
+
+ stream, err := client.CumSum(context.Background())
+ assert.Nil(t, err)
+ if sendErr := stream.Send(&pingv1.CumSumRequest{Number: 1}); sendErr !=
nil {
+ t.Fatalf("failed to send setup request before close
verification: %v", sendErr)
+ }
+
+ select {
+ case <-serverReceived:
+ case <-time.After(closeTestTimeout):
+ t.Fatal("server did not receive the setup request before close
verification")
+ }
+ close(serverReturn)
+
+ err = stream.Receive(&pingv1.CumSumResponse{})
+ assert.NotNil(t, err)
+ assert.Equal(t, triple.CodeOf(err), triple.CodeUnavailable)
+
+ done := make(chan error, 1)
+ go func() {
+ done <- stream.CloseResponse()
+ }()
+ assert.Nil(t, assertCloseReturnsPromptly(t, done, nil,
"BidiStreamForClient.CloseResponse blocked after server stopped reading"))
+ assert.Nil(t, stream.CloseRequest())
+}
+
+// TestClientStreamCloseAndReceiveAfterServerReturnsError covers a
client-stream
+// RPC where the server returns an error before the client finishes sending all
+// messages. CloseAndReceive should surface the server error and finish cleanup
+// without hanging in CloseResponse.
+func TestClientStreamCloseAndReceiveAfterServerReturnsError(t *testing.T) {
+ client := newCloseLifecyclePingClient(t, &pluggablePingServer{
+ sum: func(ctx context.Context, stream *triple.ClientStream)
(*triple.Response, error) {
+ return nil, triple.NewError(triple.CodeUnavailable,
errors.New("server returned early"))
+ },
+ })
+
+ stream, err := client.Sum(context.Background())
+ assert.Nil(t, err)
+ if err = stream.Send(&pingv1.SumRequest{Number: 1}); err != nil {
+ assert.ErrorIs(t, err, io.EOF)
+ }
+
+ done := make(chan error, 1)
+ go func() {
+ done <-
stream.CloseAndReceive(triple.NewResponse(&pingv1.SumResponse{}))
+ }()
+ err = assertCloseReturnsPromptly(t, done, nil,
"ClientStreamForClient.CloseAndReceive blocked after server returned early")
+ assert.NotNil(t, err)
+ assert.Equal(t, triple.CodeOf(err), triple.CodeUnavailable)
+}
+
+func newCloseLifecyclePingClient(t *testing.T, pingServer
pingv1connect.PingServiceHandler) pingv1connect.PingServiceClient {
+ t.Helper()
+ mux := http.NewServeMux()
+ mux.Handle(pingv1connect.NewPingServiceHandler(pingServer))
+ server := httptest.NewUnstartedServer(mux)
+ server.EnableHTTP2 = true
+ server.StartTLS()
+ t.Cleanup(server.Close)
+ return pingv1connect.NewPingServiceClient(server.Client(), server.URL)
+}
+
+// newCloseRelease returns a channel that lets tests keep the server handler
+// alive until the client-side close call has completed. The unblock function
is
+// idempotent so cleanup can safely release the handler after failures.
+func newCloseRelease() (<-chan struct{}, func()) {
+ release := make(chan struct{})
+ var once sync.Once
+ return release, func() {
+ once.Do(func() {
+ close(release)
+ })
+ }
+}
+
+// assertCloseReturnsPromptly treats a timeout as evidence that close attempted
+// to drain the response stream. On failure it releases the server first so the
+// goroutine running the close call can finish before the test exits.
+func assertCloseReturnsPromptly(t *testing.T, done <-chan error, unblock
func(), failure string) error {
+ t.Helper()
+ select {
+ case err := <-done:
+ return err
+ case <-time.After(closeTestTimeout):
+ if unblock != nil {
+ unblock()
+ err := <-done
+ assert.Nil(t, err)
+ }
+ t.Fatal(failure)
+ return nil
+ }
+}
