Respond to PR feedback.

Author: matthewdale

x/mongo/driver/topology/pool.go

@@ -319,10 +319,10 @@ func (p *pool) checkOut(ctx context.Context) (conn *connection, err error) {
 		}
 	}()
 
-	// Get in the queue for an idle connection. If queueForIdleConn returns true, it was able to
+	// Get in the queue for an idle connection. If getOrQueueForIdleConn returns true, it was able to
 	// immediately deliver an idle connection to the wantConn, so we can return the connection or
 	// error from the wantConn without waiting for "ready".
-	if delivered := p.queueForIdleConn(w); delivered {
+	if delivered := p.getOrQueueForIdleConn(w); delivered {
 		if w.err != nil {
 			if p.monitor != nil {
 				p.monitor.Event(&event.PoolEvent{
@@ -528,7 +528,11 @@ func (p *pool) clear(serviceID *primitive.ObjectID) {
 	p.generation.clear(serviceID)
 }
 
-func (p *pool) queueForIdleConn(w *wantConn) bool {
+// getOrQueueForIdleConn attempts to deliver an idle connection to the given wantConn. If there is
+// an idle connection in the idle connections stack, it pops an idle connection, delivers it to the
+// wantConn, and returns true. If there are no idle connections in the idle connections stack, it
+// adds the wantConn to the idleConnWait queue and returns false.
+func (p *pool) getOrQueueForIdleConn(w *wantConn) bool {
 	p.idleMu.Lock()
 	defer p.idleMu.Unlock()
 
@@ -836,6 +840,10 @@ func (w *wantConn) tryDeliver(conn *connection, err error) bool {
 // has been delivered already, cancel returns it with p.checkInNoEvent(). Note that the caller must
 // not hold any locks on the pool while calling cancel.
 func (w *wantConn) cancel(p *pool, err error) {
+	if err == nil {
+		panic("x/mongo/driver/topology: internal error: misuse of cancel")
+	}
+
 	w.mu.Lock()
 	if w.conn == nil && w.err == nil {
 		close(w.ready) // catch misbehavior in future delivery

x/mongo/driver/topology/pool_test.go

@@ -595,6 +595,8 @@ func TestPool(t *testing.T) {
 			err = p.disconnect(context.Background())
 			noerr(t, err)
 		})
+		// Test that if a checkOut() times out, it returns a WaitQueueTimeout error that wraps a
+		// context.DeadlineExceeded error.
 		t.Run("wait queue timeout error", func(t *testing.T) {
 			t.Parallel()
 
@@ -623,19 +625,72 @@ func TestPool(t *testing.T) {
 			assert.NotNilf(t, err, "expected a WaitQueueTimeout error")
 
 			// Assert that error received is WaitQueueTimeoutError with context deadline exceeded.
-			assert.IsType(t, WaitQueueTimeoutError{}, err, "expected a WaitQueueTimeoutError")
+			assert.IsTypef(t, WaitQueueTimeoutError{}, err, "expected a WaitQueueTimeoutError")
 			if err, ok := err.(WaitQueueTimeoutError); ok {
 				assert.Equalf(t, context.DeadlineExceeded, err.Unwrap(), "expected wrapped error to be a context.Timeout")
 			}
 
 			err = p.disconnect(context.Background())
 			noerr(t, err)
 		})
+		// Test that an indefinitely blocked checkOut() doesn't cause the wait queue to overflow
+		// if there are many other checkOut() calls that time out. This tests a scenario where a
+		// wantConnQueue may grow unbounded while a checkOut() is blocked, even if all subsequent
+		// checkOut() calls time out (due to the behavior of wantConnQueue.cleanFront()).
+		t.Run("wait queue doesn't overflow", func(t *testing.T) {
+			t.Parallel()
+
+			cleanup := make(chan struct{})
+			defer close(cleanup)
+			addr := bootstrapConnections(t, 1, func(nc net.Conn) {
+				<-cleanup
+				_ = nc.Close()
+			})
+
+			p := newPool(poolConfig{
+				Address:     address.Address(addr.String()),
+				MaxPoolSize: 1,
+			})
+			err := p.connect()
+			noerr(t, err)
+
+			// Check out the 1 connection that the pool will create.
+			c, err := p.checkOut(context.Background())
+			noerr(t, err)
+
+			// Start a goroutine that tries to check out another connection with no timeout. Expect
+			// this goroutine to block (wait in the wait queue) until the checked-out connection is
+			// checked-in. Assert that there is no error once checkOut() finally does return.
+			var wg sync.WaitGroup
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+
+				_, err := p.checkOut(context.Background())
+				noerr(t, err)
+			}()
+
+			// Run lots of check-out attempts with a low timeout and assert that each one fails with
+			// a WaitQueueTimeout error. Expect no other errors or panics.
+			for i := 0; i < 50000; i++ {
+				ctx, cancel := context.WithTimeout(context.Background(), 1*time.Microsecond)
+				_, err := p.checkOut(ctx)
+				cancel()
+				assert.NotNilf(t, err, "expected a WaitQueueTimeout error")
+				assert.IsTypef(t, WaitQueueTimeoutError{}, err, "expected a WaitQueueTimeoutError")
+			}
+
+			// Check-in the connection we checked out earlier and wait for the checkOut() goroutine
+			// to resume.
+			err = p.checkIn(c)
+			noerr(t, err)
+			wg.Wait()
+		})
 	})
 	t.Run("checkIn", func(t *testing.T) {
 		t.Parallel()
 
-		t.Run("does not return to pool twice", func(t *testing.T) {
+		t.Run("cannot return same connection to pool twice", func(t *testing.T) {
 			t.Parallel()
 
 			cleanup := make(chan struct{})