GODRIVER-3173 Add prose tests

Author: prestonvasquez

internal/integration/csot_prose_test.go

@@ -233,9 +233,6 @@ func TestCSOTProse(t *testing.T) {
 		assert.ErrorIs(mt, err, context.DeadlineExceeded, "expected a timeout error, got: %v", err)
 		assert.Equal(mt, 2, cnt, "expected bulkWrite calls: %d, got: %d", 2, cnt)
 	})
-
-	//mt.Run("pending read", func(mt *mtest.T) {
-	//})
 }
 
 func TestCSOTProse_GridFS(t *testing.T) {

internal/integration/unified/unified_spec_runner.go

@@ -122,8 +122,6 @@ func runTestFile(t *testing.T, filepath string, expectValidFail bool, opts ...*O
 			defer func() {
 				// catch panics from looking up elements and fail if it's unexpected
 				if r := recover(); r != nil {
-
-					panic(r)
 					if !expectValidFail {
 						mt.Fatal(r)
 					}

x/mongo/driver/topology/pool.go

@@ -788,7 +788,7 @@ func (p *pool) removeConnection(conn *connection, reason reason, err error) erro
 //
 // Deprecated: PendingReadTimeout is intended for internal use only and may be
 // removed or modified at any time.
-var PendingReadTimeout = 400 * time.Millisecond
+var PendingReadTimeout = 3000 * time.Millisecond
 
 // publishPendingReadStarted will log a message to the pool logger and
 // publish an event to the pool monitor if they are set.
@@ -880,34 +880,11 @@ func publishPendingReadSucceeded(pool *pool, conn *connection) {
 	}
 }
 
-// newPendingReadContext creates a new context with a deadline that is the
-// minimum of the parent context's deadline and the remaining time.
-func newPendingReadContext(parent context.Context, remainingTime time.Duration) (context.Context, context.CancelFunc) {
-	parentDeadline, hasDeadline := parent.Deadline()
-
-	calculatedDeadline := time.Now().Add(remainingTime)
-
-	if hasDeadline {
-		// Chose the earliest of the two deadlines.
-		var minDeadline time.Time
-		if calculatedDeadline.Before(parentDeadline) {
-			minDeadline = calculatedDeadline
-		} else {
-			minDeadline = parentDeadline
-		}
-
-		return context.WithDeadline(parent, minDeadline)
-	}
-
-	// If no deadline was set on the parent context, use the remaining time.
-	return context.WithTimeout(parent, remainingTime)
-}
-
 // peekConnectionAlive checks if the connection is alive by peeking at the
 // buffered reader. If the connection is closed, it will return false.
 func peekConnectionAlive(conn *connection) (int, error) {
 	// Set a very short deadline to avoid blocking.
-	if err := conn.nc.SetReadDeadline(time.Now().Add(1 * time.Nanosecond)); err != nil {
+	if err := conn.nc.SetReadDeadline(time.Now().Add(1 * time.Millisecond)); err != nil {
 		return 0, err
 	}
 
@@ -1051,6 +1028,7 @@ func awaitPendingRead(ctx context.Context, pool *pool, conn *connection) error {
 		}()
 
 		if netErr, ok := err.(net.Error); ok && !netErr.Timeout() {
+			fmt.Println(1)
 			if err := conn.close(); err != nil {
 				return err
 			}

x/mongo/driver/topology/pool_test.go

@@ -9,6 +9,7 @@ package topology
 import (
 	"context"
 	"errors"
+	"io"
 	"net"
 	"regexp"
 	"sync"
@@ -1566,61 +1567,156 @@ func TestAwaitPendingRead(t *testing.T) {
 
 		assert.EqualError(t, pendingReadError, "error discarding 3 byte message: EOF")
 	})
-	//t.Run("no remaining time with no response", func(t *testing.T) {
-	//	timeout := 10 * time.Millisecond
-
-	//	addr := bootstrapConnections(t, 1, func(nc net.Conn) {
-	//		defer func() {
-	//			_ = nc.Close()
-	//		}()
-
-	//		// Write nothing so that the 10 microseconds "non-blocking"
-	//	})
-
-	//	var pendingReadError error
-	//	monitor := &event.PoolMonitor{
-	//		Event: func(pe *event.PoolEvent) {
-	//			if pe.Type == event.ConnectionPendingReadFailed {
-	//				pendingReadError = pe.Error
-	//			}
-	//		},
-	//	}
-
-	//	p := newPool(
-	//		poolConfig{
-	//			Address:     address.Address(addr.String()),
-	//			PoolMonitor: monitor,
-	//		},
-	//	)
-	//	defer p.close(context.Background())
-	//	err := p.ready()
-	//	require.NoError(t, err)
-
-	//	conn, err := p.checkOut(context.Background())
-	//	require.NoError(t, err)
-
-	//	ctx, cancel := csot.WithTimeout(context.Background(), &timeout)
-	//	defer cancel()
-
-	//	ctx = driverutil.WithValueHasMaxTimeMS(ctx, true)
-	//	ctx = driverutil.WithRequestID(ctx, -1)
-
-	//	_, err = conn.readWireMessage(ctx)
-	//	regex := regexp.MustCompile(
-	//		`^connection\(.*\[-\d+\]\) incomplete read of message header: context deadline exceeded: read tcp 127.0.0.1:.*->127.0.0.1:.*: i\/o timeout$`,
-	//	)
-	//	assert.True(t, regex.MatchString(err.Error()), "error %q does not match pattern %q", err, regex)
-	//	err = p.checkIn(conn)
-	//	require.NoError(t, err)
-
-	//	// Wait 400ms to ensure there is no time left on the pending read.
-	//	time.Sleep(400 * time.Millisecond)
-
-	//	_, err = p.checkOut(context.Background())
-	//	require.NoError(t, err)
-
-	//	require.NoError(t, pendingReadError)
-	//})
+
+	// Need to test the case where we attempt a non-blocking read to determine if
+	// we should refresh the remaining time. In the case of the Go Driver, we do
+	// this by attempt to "pee" at 1 byte with a deadline of 1ns.
+	t.Run("connection attempts peek but fails", func(t *testing.T) {
+		timeout := 10 * time.Millisecond
+
+		// Mock a TCP listener that will write a byte sequence > 5 (to avoid errors
+		// due to size) to the TCP socket. Have the listener sleep for 2x the
+		// timeout provided to the connection AFTER writing the byte sequence. This
+		// wiill cause the connection to timeout while reading from the socket.
+		addr := bootstrapConnections(t, 1, func(nc net.Conn) {
+			defer func() {
+				_ = nc.Close()
+			}()
+
+			_, err := nc.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
+			require.NoError(t, err)
+			time.Sleep(timeout * 2)
+
+			// Write nothing so that the 1 millisecond "non-blocking" peek fails.
+		})
+
+		var pendingReadError error
+		monitor := &event.PoolMonitor{
+			Event: func(pe *event.PoolEvent) {
+				if pe.Type == event.ConnectionPendingReadFailed {
+					pendingReadError = pe.Error
+				}
+			},
+		}
+
+		p := newPool(
+			poolConfig{
+				Address:     address.Address(addr.String()),
+				PoolMonitor: monitor,
+			},
+		)
+		defer p.close(context.Background())
+		err := p.ready()
+		require.NoError(t, err)
+
+		// Check out a connection and read from the socket, causing a timeout and
+		// pinning the connection to a pending read state.
+		conn, err := p.checkOut(context.Background())
+		require.NoError(t, err)
+
+		ctx, cancel := csot.WithTimeout(context.Background(), &timeout)
+		defer cancel()
+
+		ctx = driverutil.WithValueHasMaxTimeMS(ctx, true)
+		ctx = driverutil.WithRequestID(ctx, -1)
+
+		_, err = conn.readWireMessage(ctx)
+		regex := regexp.MustCompile(
+			`^connection\(.*\[-\d+\]\) incomplete read of full message: context deadline exceeded: read tcp 127.0.0.1:.*->127.0.0.1:.*: i\/o timeout$`,
+		)
+		assert.True(t, regex.MatchString(err.Error()), "error %q does not match pattern %q", err, regex)
+
+		// Check in the connection with a pending read state. The next time this
+		// connection is checked out, it should attempt to read the pending
+		// response.
+		err = p.checkIn(conn)
+		require.NoError(t, err)
+
+		// Wait 3s to make sure there is no remaining time on the pending read
+		// state.
+		time.Sleep(3 * time.Second)
+
+		// Check out the connection again. The remaining time should be exhausted
+		// requiring us to "peek" at the connection to determine if we should
+		_, err = p.checkOut(context.Background())
+		assert.ErrorIs(t, err, io.EOF)
+		assert.ErrorIs(t, pendingReadError, io.EOF)
+	})
+
+	t.Run("connection attempts peek and succeeds", func(t *testing.T) {
+		timeout := 10 * time.Millisecond
+
+		// Mock a TCP listener that will write a byte sequence > 5 (to avoid errors
+		// due to size) to the TCP socket. Have the listener sleep for 2x the
+		// timeout provided to the connection AFTER writing the byte sequence. This
+		// wiill cause the connection to timeout while reading from the socket.
+		addr := bootstrapConnections(t, 1, func(nc net.Conn) {
+			defer func() {
+				_ = nc.Close()
+			}()
+
+			_, err := nc.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
+			require.NoError(t, err)
+			time.Sleep(timeout * 2)
+
+			// Write data that can be peeked at.
+			_, err = nc.Write([]byte{12, 0, 0, 0, 0, 0, 0, 0, 1})
+			require.NoError(t, err)
+
+		})
+
+		var pendingReadError error
+		monitor := &event.PoolMonitor{
+			Event: func(pe *event.PoolEvent) {
+				if pe.Type == event.ConnectionPendingReadFailed {
+					pendingReadError = pe.Error
+				}
+			},
+		}
+
+		p := newPool(
+			poolConfig{
+				Address:     address.Address(addr.String()),
+				PoolMonitor: monitor,
+			},
+		)
+		defer p.close(context.Background())
+		err := p.ready()
+		require.NoError(t, err)
+
+		// Check out a connection and read from the socket, causing a timeout and
+		// pinning the connection to a pending read state.
+		conn, err := p.checkOut(context.Background())
+		require.NoError(t, err)
+
+		ctx, cancel := csot.WithTimeout(context.Background(), &timeout)
+		defer cancel()
+
+		ctx = driverutil.WithValueHasMaxTimeMS(ctx, true)
+		ctx = driverutil.WithRequestID(ctx, -1)
+
+		_, err = conn.readWireMessage(ctx)
+		regex := regexp.MustCompile(
+			`^connection\(.*\[-\d+\]\) incomplete read of full message: context deadline exceeded: read tcp 127.0.0.1:.*->127.0.0.1:.*: i\/o timeout$`,
+		)
+		assert.True(t, regex.MatchString(err.Error()), "error %q does not match pattern %q", err, regex)
+
+		// Check in the connection with a pending read state. The next time this
+		// connection is checked out, it should attempt to read the pending
+		// response.
+		err = p.checkIn(conn)
+		require.NoError(t, err)
+
+		// Wait 3s to make sure there is no remaining time on the pending read
+		// state.
+		time.Sleep(3 * time.Second)
+
+		// Check out the connection again. The remaining time should be exhausted
+		// requiring us to "peek" at the connection to determine if we should
+		_, err = p.checkOut(context.Background())
+		require.NoError(t, err)
+		require.NoError(t, pendingReadError)
+	})
 }
 
 func assertConnectionsClosed(t *testing.T, dialer *dialer, count int) {