AlexStocks commented on code in PR #3466: URL: https://github.com/apache/dubbo-go/pull/3466#discussion_r3502744174
########## protocol/triple/triple_protocol/stream_close_ext_test.go: ########## @@ -0,0 +1,202 @@ +/* + * 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" +) + +// TestServerStreamCloseDoesNotDrainResponse covers the public +// ServerStreamForClient.Close path for connect-go#791-style lifecycle +// behavior. The server sends one response and then intentionally keeps the +// stream open; client-side Close should still return promptly because it is a +// cleanup operation, not a request to read the stream to EOF. +func TestServerStreamCloseDoesNotDrainResponse(t *testing.T) { + release, unblock := newCloseRelease() + t.Cleanup(unblock) + 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 + } + <-release + return nil + }, + }) + + stream, err := client.CountUp(context.Background(), triple.NewRequest(&pingv1.CountUpRequest{})) + assert.Nil(t, err) + assert.True(t, stream.Receive(&pingv1.CountUpResponse{})) Review Comment: [P1] 这里把它当作验证 close 行为的回归测试，但我在当前 PR 分支本地执行作者自己在 PR 描述里给出的命令后，这条用例会先在 `assert.True(t, stream.Receive(...))` 失败，根本还没进入 `stream.Close()` 路径。这样它不能证明本次 `CloseRead` 改动的高层行为。建议先修正这个测试的建模/初始化条件，确保它至少能稳定收第一条消息，再用它验证 close 不阻塞。 ########## protocol/triple/triple_protocol/stream_close_ext_test.go: ########## @@ -0,0 +1,202 @@ +/* + * 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" +) + +// TestServerStreamCloseDoesNotDrainResponse covers the public +// ServerStreamForClient.Close path for connect-go#791-style lifecycle +// behavior. The server sends one response and then intentionally keeps the +// stream open; client-side Close should still return promptly because it is a +// cleanup operation, not a request to read the stream to EOF. +func TestServerStreamCloseDoesNotDrainResponse(t *testing.T) { + release, unblock := newCloseRelease() + t.Cleanup(unblock) + 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 + } + <-release + return nil + }, + }) + + stream, err := client.CountUp(context.Background(), triple.NewRequest(&pingv1.CountUpRequest{})) + assert.Nil(t, err) + assert.True(t, stream.Receive(&pingv1.CountUpResponse{})) + 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 covers the public +// BidiStreamForClient.CloseResponse path. After one successful receive, the +// server keeps the response side open; closing the receive side should close +// the HTTP response body instead of draining it. +func TestBidiStreamCloseResponseDoesNotDrainResponse(t *testing.T) { + release, unblock := newCloseRelease() + t.Cleanup(unblock) + 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 + } + if err := stream.Send(&pingv1.CumSumResponse{Sum: req.Number}); err != nil { + return err + } + <-release + return nil + }, + }) + + stream, err := client.CumSum(context.Background()) + assert.Nil(t, err) + assert.Nil(t, stream.Send(&pingv1.CumSumRequest{Number: 2})) Review Comment: [P1] 这条回归测试当前在本地并不能走到 `CloseResponse()`：`stream.Send(...)` 这里直接返回 `unknown: write envelope: EOF`。既然失败点发生在 close 前，这个用例就不能作为“close 不再 drain 响应体”的验证证据。建议先把 bidi 场景下首个 request/response 往返稳定下来，再断言 `CloseResponse()` 的非阻塞行为。 ########## protocol/triple/triple_protocol/stream_close_ext_test.go: ########## @@ -0,0 +1,202 @@ +/* + * 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" +) + +// TestServerStreamCloseDoesNotDrainResponse covers the public +// ServerStreamForClient.Close path for connect-go#791-style lifecycle +// behavior. The server sends one response and then intentionally keeps the +// stream open; client-side Close should still return promptly because it is a +// cleanup operation, not a request to read the stream to EOF. +func TestServerStreamCloseDoesNotDrainResponse(t *testing.T) { + release, unblock := newCloseRelease() + t.Cleanup(unblock) + 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 + } + <-release + return nil + }, + }) + + stream, err := client.CountUp(context.Background(), triple.NewRequest(&pingv1.CountUpRequest{})) + assert.Nil(t, err) + assert.True(t, stream.Receive(&pingv1.CountUpResponse{})) + 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 covers the public +// BidiStreamForClient.CloseResponse path. After one successful receive, the +// server keeps the response side open; closing the receive side should close +// the HTTP response body instead of draining it. +func TestBidiStreamCloseResponseDoesNotDrainResponse(t *testing.T) { + release, unblock := newCloseRelease() + t.Cleanup(unblock) + 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 + } + if err := stream.Send(&pingv1.CumSumResponse{Sum: req.Number}); err != nil { + return err + } + <-release + return nil + }, + }) + + stream, err := client.CumSum(context.Background()) + assert.Nil(t, err) + assert.Nil(t, stream.Send(&pingv1.CumSumRequest{Number: 2})) + res := &pingv1.CumSumResponse{} + assert.Nil(t, stream.Receive(res)) + 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 before consuming the rest of the request stream. Once the +// client has received the server-side error, closing the receive side should be +// a local cleanup step and must not wait for additional response bytes. +func TestBidiStreamCloseResponseAfterServerStopsReading(t *testing.T) { + 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 + } + return triple.NewError(triple.CodeUnavailable, errors.New("server stopped reading")) + }, + }) + + stream, err := client.CumSum(context.Background()) + assert.Nil(t, err) + assert.Nil(t, stream.Send(&pingv1.CumSumRequest{Number: 1})) Review Comment: [P1] 同样的问题，这条用例在当前 PR 分支本地会先失败在 `stream.Send(...)`，错误是 `unknown: write envelope: EOF`，还没进入后面的 `stream.Receive(...)` / `CloseResponse()` 路径。现在这条测试验证不到 PR 想修的生命周期问题。建议把“server stopped reading”场景和 close 行为拆开：先保证 client 能稳定拿到服务端错误，再验证随后 `CloseResponse()` 不会阻塞。 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
