fix

Author: qingyang-hu

x/mongo/driver/topology/pool.go

@@ -902,8 +902,9 @@ func (p *pool) checkInNoEvent(conn *connection) error {
 	// not usable, which is not covered by the current pool events. We may need
 	// to add pool event information in the future to communicate that.
 	if conn.awaitingResponse != nil {
-		go bgRead(p, conn, *conn.awaitingResponse)
+		size := *conn.awaitingResponse
 		conn.awaitingResponse = nil
+		go bgRead(p, conn, size)
 		return nil
 	}
 

x/mongo/driver/topology/pool_test.go

@@ -12,6 +12,7 @@ import (
 	"io"
 	"net"
 	"os"
+	"regexp"
 	"sync"
 	"testing"
 	"time"
@@ -1128,9 +1129,13 @@ func TestPool(t *testing.T) {
 	t.Run("bgRead", func(t *testing.T) {
 		t.Parallel()
 
-		var gotErrs []error
+		var errCh chan error
 		BGReadCallback = func(addr string, start, read time.Time, errs []error, connClosed bool) {
-			gotErrs = errs
+			defer close(errCh)
+
+			for _, err := range errs {
+				errCh <- err
+			}
 		}
 
 		const sockPath = "./test.sock"
@@ -1140,6 +1145,8 @@ func TestPool(t *testing.T) {
 		setup := func(t *testing.T) {
 			t.Helper()
 
+			errCh = make(chan error)
+
 			var err error
 			socket, err = net.Listen("unix", sockPath)
 			noerr(t, err)
@@ -1148,30 +1155,26 @@ func TestPool(t *testing.T) {
 			t.Helper()
 
 			os.Remove(sockPath)
-			gotErrs = nil
 		}
 
 		t.Run("incomplete read of message header", func(t *testing.T) {
 			setup(t)
 			defer teardown(t)
 
 			wg := &sync.WaitGroup{}
+			wg.Add(1)
 			go func(t *testing.T) {
 				t.Helper()
 
+				defer wg.Done()
+
 				conn, err := socket.Accept()
 				noerr(t, err)
-				wg.Add(1)
-				go func(conn net.Conn) {
-					defer func() {
-						conn.Close()
-						wg.Done()
-					}()
+				defer conn.Close()
 
-					_, err = conn.Write([]byte{10, 0, 0})
-					noerr(t, err)
-					time.Sleep(1500 * time.Millisecond)
-				}(conn)
+				_, err = conn.Write([]byte{10, 0, 0})
+				noerr(t, err)
+				time.Sleep(1500 * time.Millisecond)
 			}(t)
 
 			p := newPool(
@@ -1190,36 +1193,34 @@ func TestPool(t *testing.T) {
 			ctx, cancel := csot.MakeTimeoutContext(context.Background(), time.Second)
 			defer cancel()
 			_, err = conn.readWireMessage(ctx)
-			assert.ErrorContains(t, err,
-				"incomplete read of message header: context deadline exceeded: read unix ->./test.sock: i/o timeout")
-			err = p.checkIn(conn)
-			noerr(t, err)
+			regex := regexp.MustCompile(
+				`^connection\(.*\[-\d+\]\) incomplete read of message header: context deadline exceeded: read unix .*->\.\/test.sock: i\/o timeout$`,
+			)
+			assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)
+			assert.Nil(t, conn.awaitingResponse, "conn.awaitingResponse should be nil")
 			wg.Wait()
 			p.close(context.Background())
-			assert.Len(t, gotErrs, 0)
+			close(errCh)
 		})
 		t.Run("timeout on reading the message header", func(t *testing.T) {
 			setup(t)
 			defer teardown(t)
 
 			wg := &sync.WaitGroup{}
+			wg.Add(1)
 			go func(t *testing.T) {
 				t.Helper()
 
+				defer wg.Done()
+
 				conn, err := socket.Accept()
 				noerr(t, err)
-				wg.Add(1)
-				go func(conn net.Conn) {
-					defer func() {
-						conn.Close()
-						wg.Done()
-					}()
-
-					time.Sleep(1500 * time.Millisecond)
-					_, err = conn.Write([]byte{10, 0, 0, 0, 0, 0, 0, 0})
-					noerr(t, err)
-					time.Sleep(1500 * time.Millisecond)
-				}(conn)
+				defer conn.Close()
+
+				time.Sleep(1500 * time.Millisecond)
+				_, err = conn.Write([]byte{10, 0, 0, 0, 0, 0, 0, 0})
+				noerr(t, err)
+				time.Sleep(1500 * time.Millisecond)
 			}(t)
 
 			p := newPool(
@@ -1238,42 +1239,54 @@ func TestPool(t *testing.T) {
 			ctx, cancel := csot.MakeTimeoutContext(context.Background(), time.Second)
 			defer cancel()
 			_, err = conn.readWireMessage(ctx)
-			assert.ErrorContains(t, err,
-				"incomplete read of message header: context deadline exceeded: read unix ->./test.sock: i/o timeout")
+			regex := regexp.MustCompile(
+				`^connection\(.*\[-\d+\]\) incomplete read of message header: context deadline exceeded: read unix .*->\.\/test.sock: i\/o timeout$`,
+			)
+			assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)
 			err = p.checkIn(conn)
 			noerr(t, err)
 			wg.Wait()
 			p.close(context.Background())
-			assert.Len(t, gotErrs, 1)
-			for _, err = range gotErrs {
-				assert.ErrorContains(t, err,
-					"error reading message of 6: read unix ->./test.sock: i/o timeout")
+			errs := []*regexp.Regexp{
+				regexp.MustCompile(
+					`^error reading message of 6: read unix .*->\.\/test.sock: i\/o timeout$`,
+				),
+			}
+			for i := 0; true; i++ {
+				err, ok := <-errCh
+				if !ok {
+					if i != len(errs) {
+						assert.Fail(t, "expected more errors")
+					}
+					break
+				} else if i < len(errs) {
+					assert.True(t, errs[i].MatchString(err.Error()), "mismatched err: %v", err)
+				} else {
+					assert.Fail(t, "unexpected error", "got unexpected error: %v", err)
+				}
 			}
 		})
 		t.Run("timeout on reading the full message", func(t *testing.T) {
 			setup(t)
 			defer teardown(t)
 
 			wg := &sync.WaitGroup{}
+			wg.Add(1)
 			go func(t *testing.T) {
 				t.Helper()
 
+				defer wg.Done()
+
 				conn, err := socket.Accept()
 				noerr(t, err)
-				wg.Add(1)
-				go func(conn net.Conn) {
-					defer func() {
-						conn.Close()
-						wg.Done()
-					}()
+				defer conn.Close()
 
-					_, err = conn.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
-					noerr(t, err)
-					time.Sleep(1500 * time.Millisecond)
-					_, err = conn.Write([]byte{2, 3, 4})
-					noerr(t, err)
-					time.Sleep(1500 * time.Millisecond)
-				}(conn)
+				_, err = conn.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
+				noerr(t, err)
+				time.Sleep(1500 * time.Millisecond)
+				_, err = conn.Write([]byte{2, 3, 4})
+				noerr(t, err)
+				time.Sleep(1500 * time.Millisecond)
 			}(t)
 
 			p := newPool(
@@ -1291,19 +1304,32 @@ func TestPool(t *testing.T) {
 
 			conn, err := p.checkOut(context.Background())
 			noerr(t, err)
-			ctx, cancel := csot.MakeTimeoutContext(context.Background(), time.Second)
+			ctx, cancel := csot.MakeTimeoutContext(context.Background(), 1*time.Second)
 			defer cancel()
 			_, err = conn.readWireMessage(ctx)
-			assert.ErrorContains(t, err,
-				"incomplete read of full message: context deadline exceeded: read unix ->./test.sock: i/o timeout")
+			regex := regexp.MustCompile(
+				`^connection\(.*\[-\d+\]\) incomplete read of full message: context deadline exceeded: read unix .*->\.\/test.sock: i\/o timeout$`,
+			)
+			assert.True(t, regex.MatchString(err.Error()), "mismatched err: %v", err)
 			err = p.checkIn(conn)
 			noerr(t, err)
 			wg.Wait()
 			p.close(context.Background())
-			assert.Len(t, gotErrs, 1)
-			for _, err = range gotErrs {
-				assert.ErrorContains(t, err,
-					"error reading message of 3: EOF")
+			errs := []string{
+				"error reading message of 3: EOF",
+			}
+			for i := 0; true; i++ {
+				err, ok := <-errCh
+				if !ok {
+					if i != len(errs) {
+						assert.Fail(t, "expected more errors")
+					}
+					break
+				} else if i < len(errs) {
+					assert.EqualError(t, err, errs[i], "mismatched err: %v", err)
+				} else {
+					assert.Fail(t, "unexpected error", "got unexpected error: %v", err)
+				}
 			}
 		})
 	})