Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Debug flaky TestStreamForServer test #627

Merged
merged 6 commits into from
Nov 10, 2023
Merged

Debug flaky TestStreamForServer test #627

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
08555a9
Debug falky TestStreamForServer test
emcfarlane Nov 6, 2023
c0c4d63
Fix lint
emcfarlane Nov 6, 2023
3b26150
Avoid nil first write errors
emcfarlane Nov 9, 2023
fb24122
Use atomic over once closure
emcfarlane Nov 9, 2023
d396735
Fix doc
emcfarlane Nov 9, 2023
fed7140
Merge branch 'main' into ed/debug-stream4server
emcfarlane Nov 10, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 13 additions & 10 deletions connect_ext_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1511,7 +1511,8 @@ func TestBidiStreamServerSendsFirstMessage(t *testing.T) {

func TestStreamForServer(t *testing.T) {
t.Parallel()
newPingClient := func(pingServer pingv1connect.PingServiceHandler) pingv1connect.PingServiceClient {
newPingClient := func(t *testing.T, pingServer pingv1connect.PingServiceHandler) pingv1connect.PingServiceClient {
t.Helper()
Comment on lines +1514 to +1515
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think you can undo all of the changes in this file, right? Looks like the addition of t here is unused, and this appears to be the only change in here.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The servers close are matched to the clients which is nice. The other change in this file is the two Receives for the test cases that Send a non nil message.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As in server := memhttptest.NewServer(t, mux) takes the correct subtests *testing.T rather than the parent.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, okay. That was a bit subtle due to the shadowing of the name t.

mux := http.NewServeMux()
mux.Handle(pingv1connect.NewPingServiceHandler(pingServer))
server := memhttptest.NewServer(t, mux)
Expand All @@ -1523,7 +1524,7 @@ func TestStreamForServer(t *testing.T) {
}
t.Run("not-proto-message", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
cumSum: func(ctx context.Context, stream *connect.BidiStream[pingv1.CumSumRequest, pingv1.CumSumResponse]) error {
return stream.Conn().Send("foobar")
},
Expand All @@ -1537,7 +1538,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("nil-message", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
cumSum: func(ctx context.Context, stream *connect.BidiStream[pingv1.CumSumRequest, pingv1.CumSumResponse]) error {
return stream.Send(nil)
},
Expand All @@ -1551,7 +1552,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("get-spec", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
cumSum: func(ctx context.Context, stream *connect.BidiStream[pingv1.CumSumRequest, pingv1.CumSumResponse]) error {
assert.Equal(t, stream.Spec().StreamType, connect.StreamTypeBidi)
assert.Equal(t, stream.Spec().Procedure, pingv1connect.PingServiceCumSumProcedure)
Expand All @@ -1565,7 +1566,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("server-stream", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
countUp: func(ctx context.Context, req *connect.Request[pingv1.CountUpRequest], stream *connect.ServerStream[pingv1.CountUpResponse]) error {
assert.Equal(t, stream.Conn().Spec().StreamType, connect.StreamTypeServer)
assert.Equal(t, stream.Conn().Spec().Procedure, pingv1connect.PingServiceCountUpProcedure)
Expand All @@ -1581,7 +1582,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("server-stream-send", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
countUp: func(ctx context.Context, req *connect.Request[pingv1.CountUpRequest], stream *connect.ServerStream[pingv1.CountUpResponse]) error {
assert.Nil(t, stream.Send(&pingv1.CountUpResponse{Number: 1}))
return nil
Expand All @@ -1597,7 +1598,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("server-stream-send-nil", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
countUp: func(ctx context.Context, req *connect.Request[pingv1.CountUpRequest], stream *connect.ServerStream[pingv1.CountUpResponse]) error {
stream.ResponseHeader().Set("foo", "bar")
stream.ResponseTrailer().Set("bas", "blah")
Expand All @@ -1618,7 +1619,7 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("client-stream", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
sum: func(ctx context.Context, stream *connect.ClientStream[pingv1.SumRequest]) (*connect.Response[pingv1.SumResponse], error) {
assert.Equal(t, stream.Spec().StreamType, connect.StreamTypeClient)
assert.Equal(t, stream.Spec().Procedure, pingv1connect.PingServiceSumProcedure)
Expand All @@ -1639,8 +1640,9 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("client-stream-conn", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
sum: func(ctx context.Context, stream *connect.ClientStream[pingv1.SumRequest]) (*connect.Response[pingv1.SumResponse], error) {
assert.True(t, stream.Receive())
assert.NotNil(t, stream.Conn().Send("not-proto"))
return connect.NewResponse(&pingv1.SumResponse{}), nil
},
Expand All @@ -1653,8 +1655,9 @@ func TestStreamForServer(t *testing.T) {
})
t.Run("client-stream-send-msg", func(t *testing.T) {
t.Parallel()
client := newPingClient(&pluggablePingServer{
client := newPingClient(t, &pluggablePingServer{
sum: func(ctx context.Context, stream *connect.ClientStream[pingv1.SumRequest]) (*connect.Response[pingv1.SumResponse], error) {
assert.True(t, stream.Receive())
assert.Nil(t, stream.Conn().Send(&pingv1.SumResponse{Sum: 2}))
return connect.NewResponse(&pingv1.SumResponse{}), nil
},
Expand Down
23 changes: 15 additions & 8 deletions duplex_http_call.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ import (
"io"
"net/http"
"net/url"
"sync"
"sync/atomic"
)

// duplexHTTPCall is a full-duplex stream between the client and server. The
Expand All @@ -42,9 +42,9 @@ type duplexHTTPCall struct {
requestBodyReader *io.PipeReader
requestBodyWriter *io.PipeWriter

// sendRequestOnce ensures we only send the request once.
sendRequestOnce sync.Once
request *http.Request
// requestSent ensures we only send the request once.
requestSent atomic.Bool
request *http.Request

// responseReady is closed when the response is ready or when the request
// fails. Any error on request initialisation will be set on the
Expand Down Expand Up @@ -96,11 +96,16 @@ func newDuplexHTTPCall(

// Write to the request body.
func (d *duplexHTTPCall) Write(data []byte) (int, error) {
d.ensureRequestMade()
isFirst := d.ensureRequestMade()
// Before we send any data, check if the context has been canceled.
if err := d.ctx.Err(); err != nil {
return 0, wrapIfContextError(err)
}
if isFirst && data == nil {
// On first write a nil Send is used to send request headers. Avoid
// writing a zero-length payload to avoid superfluous errors with close.
return 0, nil
}
// It's safe to write to this side of the pipe while net/http concurrently
// reads from the other side.
bytesWritten, err := d.requestBodyWriter.Write(data)
Expand Down Expand Up @@ -229,10 +234,12 @@ func (d *duplexHTTPCall) BlockUntilResponseReady() error {
// ensureRequestMade sends the request headers and starts the response stream.
// It is not safe to call this concurrently. Write and CloseWrite call this but
// ensure that they're not called concurrently.
func (d *duplexHTTPCall) ensureRequestMade() {
d.sendRequestOnce.Do(func() {
func (d *duplexHTTPCall) ensureRequestMade() (isFirst bool) {
if d.requestSent.CompareAndSwap(false, true) {
go d.makeRequest()
})
return true
}
return false
}

func (d *duplexHTTPCall) makeRequest() {
Expand Down