GODRIVER-3173 Organize prose tests

Author: prestonvasquez

event/monitoring.go

@@ -75,20 +75,20 @@ const (
 
 // strings for pool command monitoring types
 const (
-	ConnectionPoolCreated          = "ConnectionPoolCreated"
-	ConnectionPoolReady            = "ConnectionPoolReady"
-	ConnectionPoolCleared          = "ConnectionPoolCleared"
-	ConnectionPoolClosed           = "ConnectionPoolClosed"
-	ConnectionCreated              = "ConnectionCreated"
-	ConnectionReady                = "ConnectionReady"
-	ConnectionClosed               = "ConnectionClosed"
-	ConnectionCheckOutStarted      = "ConnectionCheckOutStarted"
-	ConnectionCheckOutFailed       = "ConnectionCheckOutFailed"
-	ConnectionCheckedOut           = "ConnectionCheckedOut"
-	ConnectionCheckedIn            = "ConnectionCheckedIn"
-	ConnectionPendingReadStarted   = "ConnectionPendingReadStarted"
-	ConnectionPendingReadSucceeded = "ConnectionPendingReadSucceeded"
-	ConnectionPendingReadFailed    = "ConnectionPendingReadFailed"
+	ConnectionPoolCreated              = "ConnectionPoolCreated"
+	ConnectionPoolReady                = "ConnectionPoolReady"
+	ConnectionPoolCleared              = "ConnectionPoolCleared"
+	ConnectionPoolClosed               = "ConnectionPoolClosed"
+	ConnectionCreated                  = "ConnectionCreated"
+	ConnectionReady                    = "ConnectionReady"
+	ConnectionClosed                   = "ConnectionClosed"
+	ConnectionCheckOutStarted          = "ConnectionCheckOutStarted"
+	ConnectionCheckOutFailed           = "ConnectionCheckOutFailed"
+	ConnectionCheckedOut               = "ConnectionCheckedOut"
+	ConnectionCheckedIn                = "ConnectionCheckedIn"
+	ConnectionPendingResponseStarted   = "ConnectionPendingResponseStarted"
+	ConnectionPendingResponseSucceeded = "ConnectionPendingResponseSucceeded"
+	ConnectionPendingResponseFailed    = "ConnectionPendingResponseFailed"
 )
 
 // MonitorPoolOptions contains pool options as formatted in pool events

internal/integration/mtest/mongotest.go

@@ -662,9 +662,9 @@ func (t *T) createTestClient() {
 					atomic.AddInt64(&t.connsCheckedOut, 1)
 				case event.ConnectionCheckedIn:
 					atomic.AddInt64(&t.connsCheckedOut, -1)
-				case event.ConnectionPendingReadStarted:
+				case event.ConnectionPendingResponseStarted:
 					atomic.AddInt64(&t.connPendingReadStarted, 1)
-				case event.ConnectionPendingReadSucceeded:
+				case event.ConnectionPendingResponseSucceeded:
 					atomic.AddInt64(&t.connPendingReadSucceeded, 1)
 				case event.ConnectionCheckOutFailed:
 					atomic.AddInt64(&t.connPendingReadFailed, 1)

internal/integration/unified/event.go

@@ -121,11 +121,11 @@ func monitoringEventTypeFromPoolEvent(evt *event.PoolEvent) monitoringEventType
 		return connectionCheckedOutEvent
 	case event.ConnectionCheckedIn:
 		return connectionCheckedInEvent
-	case event.ConnectionPendingReadStarted:
+	case event.ConnectionPendingResponseStarted:
 		return connectionPendingReadStarted
-	case event.ConnectionPendingReadSucceeded:
+	case event.ConnectionPendingResponseSucceeded:
 		return connectionPendingReadSucceeded
-	case event.ConnectionPendingReadFailed:
+	case event.ConnectionPendingResponseFailed:
 		return connectionPendingReadFailed
 	default:
 		return ""

internal/integration/unified/event_verification.go

@@ -363,15 +363,15 @@ func verifyCMAPEvents(client *clientEntity, expectedEvents *expectedEvents) erro
 				return newEventVerificationError(idx, client, "failed to get next pool event: %v", err.Error())
 			}
 		case evt.ConnectionPendingReadStarted != nil:
-			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingReadStarted); err != nil {
+			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingResponseStarted); err != nil {
 				return newEventVerificationError(idx, client, "failed to get next pool event: %v", err.Error())
 			}
 		case evt.ConnectionPendingreadSucceeded != nil:
-			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingReadSucceeded); err != nil {
+			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingResponseSucceeded); err != nil {
 				return newEventVerificationError(idx, client, "failed to get next pool event: %v", err.Error())
 			}
 		case evt.ConnectionPendingReadFailed != nil:
-			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingReadFailed); err != nil {
+			if _, pooled, err = getNextPoolEvent(pooled, event.ConnectionPendingResponseFailed); err != nil {
 				return newEventVerificationError(idx, client, "failed to get next pool event: %v", err.Error())
 			}
 		case evt.PoolClearedEvent != nil:

testdata/specifications

@@ -9,13 +9,19 @@ package topology
 import (
 	"context"
 	"errors"
+	"io"
 	"net"
+	"regexp"
+	"sync"
 	"testing"
 	"time"
 
 	"go.mongodb.org/mongo-driver/v2/event"
 	"go.mongodb.org/mongo-driver/v2/internal/assert"
+	"go.mongodb.org/mongo-driver/v2/internal/csot"
+	"go.mongodb.org/mongo-driver/v2/internal/driverutil"
 	"go.mongodb.org/mongo-driver/v2/internal/require"
+	"go.mongodb.org/mongo-driver/v2/mongo/address"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/operation"
 )
 
@@ -263,6 +269,201 @@ func TestCMAPProse(t *testing.T) {
 			})
 		})
 	})
+
+	// 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) {
+		const requestID = int32(-1)
+		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.
+		})
+
+		poolEventsByType := make(map[string][]event.PoolEvent)
+		poolEventsByTypeMu := &sync.Mutex{}
+
+		monitor := &event.PoolMonitor{
+			Event: func(pe *event.PoolEvent) {
+				poolEventsByTypeMu.Lock()
+				poolEventsByType[pe.Type] = append(poolEventsByType[pe.Type], *pe)
+				poolEventsByTypeMu.Unlock()
+			},
+		}
+
+		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, requestID)
+
+		_, 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)
+
+		// There should be 1 ConnectionPendingResponseStarted event.
+		started := poolEventsByType[event.ConnectionPendingResponseStarted]
+		require.Len(t, started, 1)
+
+		assert.Equal(t, addr.String(), started[0].Address)
+		assert.Equal(t, conn.driverConnectionID, started[0].ConnectionID)
+		assert.Equal(t, requestID, started[0].RequestID)
+
+		// There should be 1 ConnectionPendingResponseFailed event.
+		failed := poolEventsByType[event.ConnectionPendingResponseFailed]
+		require.Len(t, failed, 1)
+
+		assert.Equal(t, addr.String(), failed[0].Address)
+		assert.Equal(t, conn.driverConnectionID, failed[0].ConnectionID)
+		assert.Equal(t, requestID, failed[0].RequestID)
+		assert.Equal(t, io.EOF.Error(), failed[0].Reason)
+		assert.ErrorIs(t, failed[0].Error, io.EOF)
+		assert.Equal(t, time.Duration(0), failed[0].RemainingTime)
+
+		// There should be 0 ConnectionPendingResponseSucceeded event.
+		require.Len(t, poolEventsByType[event.ConnectionPendingResponseSucceeded], 0)
+	})
+
+	t.Run("connection attempts peek and succeeds", func(t *testing.T) {
+		const requestID = int32(-1)
+		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)
+
+		})
+
+		poolEventsByType := make(map[string][]event.PoolEvent)
+		poolEventsByTypeMu := &sync.Mutex{}
+
+		monitor := &event.PoolMonitor{
+			Event: func(pe *event.PoolEvent) {
+				poolEventsByTypeMu.Lock()
+				poolEventsByType[pe.Type] = append(poolEventsByType[pe.Type], *pe)
+				poolEventsByTypeMu.Unlock()
+			},
+		}
+
+		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, requestID)
+
+		_, 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)
+
+		// There should be 1 ConnectionPendingResponseStarted event.
+		started := poolEventsByType[event.ConnectionPendingResponseStarted]
+		require.Len(t, started, 1)
+
+		assert.Equal(t, addr.String(), started[0].Address)
+		assert.Equal(t, conn.driverConnectionID, started[0].ConnectionID)
+		assert.Equal(t, requestID, started[0].RequestID)
+
+		// There should be 0 ConnectionPendingResponseFailed event.
+		require.Len(t, poolEventsByType[event.ConnectionPendingResponseFailed], 0)
+
+		// There should be 1 ConnectionPendingResponseSucceeded event.
+		succeeded := poolEventsByType[event.ConnectionPendingResponseSucceeded]
+		require.Len(t, succeeded, 1)
+
+		assert.Equal(t, addr.String(), succeeded[0].Address)
+		assert.Equal(t, conn.driverConnectionID, succeeded[0].ConnectionID)
+		assert.Greater(t, int(succeeded[0].Duration), 0)
+	})
 }
 
 func createTestPool(t *testing.T, cfg poolConfig, opts ...ConnectionOption) *pool {

x/mongo/driver/topology/cmap_prose_test.go

@@ -811,7 +811,7 @@ func publishPendingReadStarted(pool *pool, conn *connection) {
 	// publish an event to the pool monitor if it is set.
 	if pool.monitor != nil {
 		event := &event.PoolEvent{
-			Type:         event.ConnectionPendingReadStarted,
+			Type:         event.ConnectionPendingResponseStarted,
 			Address:      pool.address.String(),
 			ConnectionID: conn.driverConnectionID,
 			RequestID:    prs.requestID,
@@ -840,7 +840,7 @@ func publishPendingReadFailed(pool *pool, conn *connection, err error) {
 
 	if pool.monitor != nil {
 		event := &event.PoolEvent{
-			Type:         event.ConnectionPendingReadFailed,
+			Type:         event.ConnectionPendingResponseFailed,
 			Address:      pool.address.String(),
 			ConnectionID: conn.driverConnectionID,
 			RequestID:    prs.requestID,
@@ -853,7 +853,7 @@ func publishPendingReadFailed(pool *pool, conn *connection, err error) {
 	}
 }
 
-func publishPendingReadSucceeded(pool *pool, conn *connection) {
+func publishPendingReadSucceeded(pool *pool, conn *connection, dur time.Duration) {
 	prs := conn.pendingReadState
 	if prs == nil {
 		return
@@ -870,10 +870,10 @@ func publishPendingReadSucceeded(pool *pool, conn *connection) {
 
 	if pool.monitor != nil {
 		event := &event.PoolEvent{
-			Type:         event.ConnectionPendingReadSucceeded,
+			Type:         event.ConnectionPendingResponseSucceeded,
 			Address:      pool.address.String(),
 			ConnectionID: conn.driverConnectionID,
-			//Duration:     time.Since(st),
+			Duration:     dur,
 		}
 
 		pool.monitor.Event(event)
@@ -1002,6 +1002,7 @@ func awaitPendingRead(ctx context.Context, pool *pool, conn *connection) error {
 		bytesRead        int
 	)
 
+	st := time.Now()
 	if remainingTime <= 0 {
 		// If there is no remaining time, we can just peek at the connection to check
 		// aliveness. In such cases, we don't want to close the connection.
@@ -1011,6 +1012,7 @@ func awaitPendingRead(ctx context.Context, pool *pool, conn *connection) error {
 	}
 
 	endTime := time.Now()
+	endDuration := time.Since(st)
 
 	if err != nil {
 		// No matter what happens, always check the connection back into the
@@ -1052,7 +1054,7 @@ func awaitPendingRead(ctx context.Context, pool *pool, conn *connection) error {
 		return err
 	}
 
-	publishPendingReadSucceeded(pool, conn)
+	publishPendingReadSucceeded(pool, conn, endDuration)
 
 	conn.pendingReadState = nil