GODRIVER-2038 Make all connections in the background using connectTimeoutMS.

Author: matthewdale

data/connection-monitoring-and-pooling/wait-queue-timeout.json

@@ -4,7 +4,7 @@
   "description": "must aggressively timeout threads enqueued longer than waitQueueTimeoutMS",
   "poolOptions": {
     "maxPoolSize": 1,
-    "waitQueueTimeoutMS": 20
+    "waitQueueTimeoutMS": 50
   },
   "operations": [
     {

data/connection-monitoring-and-pooling/wait-queue-timeout.yml

@@ -3,7 +3,7 @@ style: unit
 description: must aggressively timeout threads enqueued longer than waitQueueTimeoutMS
 poolOptions:
   maxPoolSize: 1
-  waitQueueTimeoutMS: 20
+  waitQueueTimeoutMS: 50
 operations:
   # Check out only possible connection
   - name: checkOut

mongo/integration/sdam_error_handling_test.go

@@ -47,46 +47,7 @@ func TestSDAMErrorHandling(t *testing.T) {
 	// blockConnection and appName.
 	mt.RunOpts("before handshake completes", baseMtOpts().Auth(true).MinServerVersion("4.4"), func(mt *mtest.T) {
 		mt.RunOpts("network errors", noClientOpts, func(mt *mtest.T) {
-			mt.Run("pool not cleared on operation-scoped network timeout", func(mt *mtest.T) {
-				// Assert that the pool is not cleared when a connection created by an application
-				// operation thread encounters an operation timeout during handshaking. Unlike the
-				// non-timeout test below, we only test connections created in the foreground for
-				// timeouts because connections created by the pool maintenance routine can't be
-				// timed out using a context.
-
-				appName := "authOperationTimeoutTest"
-				// Set failpoint on saslContinue instead of saslStart because saslStart isn't done when using
-				// speculative auth.
-				mt.SetFailPoint(mtest.FailPoint{
-					ConfigureFailPoint: "failCommand",
-					Mode: mtest.FailPointMode{
-						Times: 1,
-					},
-					Data: mtest.FailPointData{
-						FailCommands:    []string{"saslContinue"},
-						BlockConnection: true,
-						BlockTimeMS:     150,
-						AppName:         appName,
-					},
-				})
-
-				// Reset the client with the appName specified in the failpoint and the pool monitor.
-				tpm := newTestPoolMonitor()
-				mt.ResetClient(baseClientOpts().SetAppName(appName).SetPoolMonitor(tpm.PoolMonitor))
-
-				// Use a context with a 100ms timeout so that the saslContinue delay of 150ms causes
-				// an operation-scoped context timeout (i.e. a timeout not caused by a client timeout
-				// like connectTimeoutMS or socketTimeoutMS).
-				timeoutCtx, cancel := context.WithTimeout(mtest.Background, 100*time.Millisecond)
-				defer cancel()
-				_, err := mt.Coll.InsertOne(timeoutCtx, bson.D{{"test", 1}})
-				assert.NotNil(mt, err, "expected InsertOne error, got nil")
-				assert.True(mt, mongo.IsTimeout(err), "expected timeout error, got %v", err)
-				assert.True(mt, mongo.IsNetworkError(err), "expected network error, got %v", err)
-				assert.False(mt, tpm.IsPoolCleared(), "expected pool not to be cleared but was cleared")
-			})
-
-			mt.Run("pool cleared on non-operation-scoped network timeout", func(mt *mtest.T) {
+			mt.Run("pool cleared on network timeout", func(mt *mtest.T) {
 				// Assert that the pool is cleared when a connection created by an application
 				// operation thread encounters a timeout caused by connectTimeoutMS during
 				// handshaking.

x/mongo/driver/topology/CMAP_spec_test.go

@@ -119,6 +119,9 @@ func runCMAPTest(t *testing.T, testFileName string) {
 	s.connectionstate = connected
 	err = s.pool.connect()
 	testHelpers.RequireNil(t, err, "error connecting connection pool: %v", err)
+	defer func() {
+		_ = s.pool.disconnect(context.Background())
+	}()
 
 	for _, op := range test.Operations {
 		if tempErr := runOperation(t, op, testInfo, s, test.PoolOptions.WaitQueueTimeoutMS); tempErr != nil {

x/mongo/driver/topology/cmap_prose_test.go

@@ -52,13 +52,13 @@ func TestCMAPProse(t *testing.T) {
 			assert.Equal(t, numClosed, len(closed), "expected %d closed events, got %d", numClosed, len(closed))
 
 			netCount := numCreated - numClosed
-			assert.Equal(t, netCount, len(p.opened), "expected %d connections in opened map, got %d", netCount,
-				len(p.opened))
+			assert.Equal(t, netCount, p.totalConnectionCount(), "expected %d connections in opened map, got %d", netCount,
+				p.totalConnectionCount())
 		}
 
-		t.Run("get", func(t *testing.T) {
+		t.Run("checkOut", func(t *testing.T) {
 			t.Run("errored connection exists in pool", func(t *testing.T) {
-				// If a connection is created as part of minPoolSize maintenance and errors while connecting, get()
+				// If a connection is created as part of minPoolSize maintenance and errors while connecting, checkOut()
 				// should report that error and publish an event.
 				clearEvents()
 
@@ -76,20 +76,19 @@ func TestCMAPProse(t *testing.T) {
 				}
 				pool := createTestPool(t, cfg, connOpts...)
 
-				_, err := pool.get(context.Background())
-				assert.NotNil(t, err, "expected get() error, got nil")
+				_, err := pool.checkOut(context.Background())
+				assert.NotNil(t, err, "expected checkOut() error, got nil")
 
-				// If the connection doesn't finish connecting before resourcePool gives it back, the error will be
-				// detected by pool.get and result in a created/closed count of 1. If it does finish connecting, the
-				// error will be detected by resourcePool, which will return nil. Then, pool will try to create a new
-				// connection, which will also error. This process will result in a created/closed count of 2.
 				assert.True(t, len(created) == 1 || len(created) == 2, "expected 1 or 2 opened events, got %d", len(created))
 				assert.True(t, len(closed) == 1 || len(closed) == 2, "expected 1 or 2 closed events, got %d", len(closed))
+
+				_ = pool.disconnect(context.Background())
 				netCount := len(created) - len(closed)
 				assert.Equal(t, 0, netCount, "expected net connection count to be 0, got %d", netCount)
 			})
 			t.Run("pool is empty", func(t *testing.T) {
-				// If a new connection is created during get(), get() should report that error and publish an event.
+				// If a checkOut() has to create a new connection and that connection encounters an
+				// error while connecting, checkOut() should return that error and publish an event.
 				clearEvents()
 
 				var dialer DialerFunc = func(context.Context, string, string) (net.Conn, error) {
@@ -103,13 +102,16 @@ func TestCMAPProse(t *testing.T) {
 					}),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				_, err := pool.get(context.Background())
-				assert.NotNil(t, err, "expected get() error, got nil")
+				_, err := pool.checkOut(context.Background())
+				assert.NotNil(t, err, "expected checkOut() error, got nil")
 				assertConnectionCounts(t, pool, 1, 1)
 			})
 		})
-		t.Run("put", func(t *testing.T) {
+		t.Run("checkIn", func(t *testing.T) {
 			t.Run("errored connection", func(t *testing.T) {
 				// If the connection being returned to the pool encountered a network error, it should be removed from
 				// the pool and an event should be published.
@@ -124,16 +126,19 @@ func TestCMAPProse(t *testing.T) {
 					WithDialer(func(Dialer) Dialer { return dialer }),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				conn, err := pool.get(context.Background())
-				assert.Nil(t, err, "get error: %v", err)
+				conn, err := pool.checkOut(context.Background())
+				assert.Nil(t, err, "checkOut() error: %v", err)
 
 				// Force a network error by writing to the connection.
 				err = conn.writeWireMessage(context.Background(), nil)
 				assert.NotNil(t, err, "expected writeWireMessage error, got nil")
 
-				err = pool.put(conn)
-				assert.Nil(t, err, "put error: %v", err)
+				err = pool.checkIn(conn)
+				assert.Nil(t, err, "checkIn() error: %v", err)
 
 				assertConnectionCounts(t, pool, 1, 1)
 				evt := <-closed
@@ -157,16 +162,19 @@ func TestCMAPProse(t *testing.T) {
 					WithIdleTimeout(func(time.Duration) time.Duration { return 1 * time.Second }),
 				}
 				pool := createTestPool(t, getConfig(), connOpts...)
+				defer func() {
+					_ = pool.disconnect(context.Background())
+				}()
 
-				conn, err := pool.get(context.Background())
-				assert.Nil(t, err, "get error: %v", err)
+				conn, err := pool.checkOut(context.Background())
+				assert.Nil(t, err, "checkOut() error: %v", err)
 
 				// Set the idleDeadline to a time in the past to simulate expiration.
 				pastTime := time.Now().Add(-10 * time.Second)
 				conn.idleDeadline.Store(pastTime)
 
-				err = pool.put(conn)
-				assert.Nil(t, err, "put error: %v", err)
+				err = pool.checkIn(conn)
+				assert.Nil(t, err, "checkIn() error: %v", err)
 
 				assertConnectionCounts(t, pool, 1, 1)
 				evt := <-closed
@@ -189,10 +197,10 @@ func TestCMAPProse(t *testing.T) {
 				conns := checkoutConnections(t, pool, numConns)
 				assertConnectionCounts(t, pool, numConns, 0)
 
-				// Return all connections to the pool and assert that none were closed by put().
+				// Return all connections to the pool and assert that none were closed by checkIn().
 				for i, c := range conns {
-					err := pool.put(c)
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(c)
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				assertConnectionCounts(t, pool, numConns, 0)
 
@@ -223,8 +231,8 @@ func TestCMAPProse(t *testing.T) {
 
 				// Only return 2 of the connection.
 				for i := 0; i < 2; i++ {
-					err := pool.put(conns[i])
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(conns[i])
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				conns = conns[2:]
 				assertConnectionCounts(t, pool, numConns, 0)
@@ -237,8 +245,8 @@ func TestCMAPProse(t *testing.T) {
 				// Return the remaining connections and assert that the closed event count does not increase because
 				// these connections have already been closed.
 				for i, c := range conns {
-					err := pool.put(c)
-					assert.Nil(t, err, "put error at index %d: %v", i, err)
+					err := pool.checkIn(c)
+					assert.Nil(t, err, "checkIn() error at index %d: %v", i, err)
 				}
 				assertConnectionCounts(t, pool, numConns, numConns)
 
@@ -269,8 +277,8 @@ func checkoutConnections(t *testing.T, p *pool, numConns int) []*connection {
 	conns := make([]*connection, 0, numConns)
 
 	for i := 0; i < numConns; i++ {
-		conn, err := p.get(context.Background())
-		assert.Nil(t, err, "get error at index %d: %v", i, err)
+		conn, err := p.checkOut(context.Background())
+		assert.Nil(t, err, "checkOut() error at index %d: %v", i, err)
 		conns = append(conns, conn)
 	}
 

x/mongo/driver/topology/connection.go

@@ -18,7 +18,6 @@ import (
 	"sync/atomic"
 	"time"
 
-	"go.mongodb.org/mongo-driver/event"
 	"go.mongodb.org/mongo-driver/internal"
 	"go.mongodb.org/mongo-driver/mongo/address"
 	"go.mongodb.org/mongo-driver/mongo/description"
@@ -46,7 +45,6 @@ type connection struct {
 	idleDeadline         atomic.Value // Stores a time.Time
 	readTimeout          time.Duration
 	writeTimeout         time.Duration
-	descMu               sync.RWMutex // Guards desc. TODO: Remove with or after GODRIVER-2038.
 	desc                 description.Server
 	helloRTT             time.Duration
 	compressor           wiremessage.CompressorID
@@ -64,11 +62,9 @@ type connection struct {
 	cancellationListener cancellationListener
 
 	// pool related fields
-	pool         *pool
-	poolID       uint64
-	generation   uint64
-	expireReason string
-	poolMonitor  *event.PoolMonitor
+	pool       *pool
+	poolID     uint64
+	generation uint64
 }
 
 // newConnection handles the creation of a connection. It does not connect the connection.
@@ -90,7 +86,6 @@ func newConnection(addr address.Address, opts ...ConnectionOption) (*connection,
 		config:               cfg,
 		connectContextMade:   make(chan struct{}),
 		cancellationListener: internal.NewCancellationListener(),
-		poolMonitor:          cfg.poolMonitor,
 	}
 	// Connections to non-load balanced deployments should eagerly set the generation numbers so errors encountered
 	// at any point during connection establishment can be processed without the connection being considered stale.
@@ -212,13 +207,6 @@ func (c *connection) connect(ctx context.Context) {
 	// running hello and authentication is handled by a handshaker on the configuration instance.
 	handshaker := c.config.handshaker
 	if handshaker == nil {
-		if c.poolMonitor != nil {
-			c.poolMonitor.Event(&event.PoolEvent{
-				Type:         event.ConnectionReady,
-				Address:      c.addr.String(),
-				ConnectionID: c.poolID,
-			})
-		}
 		return
 	}
 
@@ -229,9 +217,7 @@ func (c *connection) connect(ctx context.Context) {
 	if err == nil {
 		// We only need to retain the Description field as the connection's description. The authentication-related
 		// fields in handshakeInfo are tracked by the handshaker if necessary.
-		c.descMu.Lock()
 		c.desc = handshakeInfo.Description
-		c.descMu.Unlock()
 		c.helloRTT = time.Since(handshakeStartTime)
 
 		// If the application has indicated that the cluster is load balanced, ensure the server has included serviceId
@@ -287,13 +273,6 @@ func (c *connection) connect(ctx context.Context) {
 			}
 		}
 	}
-	if c.poolMonitor != nil {
-		c.poolMonitor.Event(&event.PoolEvent{
-			Type:         event.ConnectionReady,
-			Address:      c.addr.String(),
-			ConnectionID: c.poolID,
-		})
-	}
 }
 
 func (c *connection) wait() error {
@@ -707,7 +686,7 @@ func (c *Connection) Expire() error {
 }
 
 func (c *Connection) cleanupReferences() error {
-	err := c.pool.put(c.connection)
+	err := c.pool.checkIn(c.connection)
 	if c.cleanupPoolFn != nil {
 		c.cleanupPoolFn()
 		c.cleanupPoolFn = nil

x/mongo/driver/topology/connection_options.go

@@ -46,7 +46,6 @@ type connectionConfig struct {
 	handshaker               Handshaker
 	idleTimeout              time.Duration
 	cmdMonitor               *event.CommandMonitor
-	poolMonitor              *event.PoolMonitor
 	readTimeout              time.Duration
 	writeTimeout             time.Duration
 	tlsConfig                *tls.Config
@@ -173,14 +172,6 @@ func WithMonitor(fn func(*event.CommandMonitor) *event.CommandMonitor) Connectio
 	}
 }
 
-// withPoolMonitor configures a event for connection monitoring.
-func withPoolMonitor(fn func(*event.PoolMonitor) *event.PoolMonitor) ConnectionOption {
-	return func(c *connectionConfig) error {
-		c.poolMonitor = fn(c.poolMonitor)
-		return nil
-	}
-}
-
 // WithZlibLevel sets the zLib compression level.
 func WithZlibLevel(fn func(*int) *int) ConnectionOption {
 	return func(c *connectionConfig) error {