Run goleak tests on replica set and sharded. Run change streams. Add comments.

Author: matthewdale

.evergreen/config.yml

@@ -1431,7 +1431,8 @@ tasks:
       - func: bootstrap-mongo-orchestration
       - func: run-docker-test
 
-  - name: test-goroutine-leaks
+  - name: test-goroutine-leaks-replicaset
+    tags: ["goleak"]
     commands:
       - func: bootstrap-mongo-orchestration
         vars:
@@ -1440,6 +1441,16 @@ tasks:
           SSL: "nossl"
       - func: run-goleak-test
 
+  - name: test-goroutine-leaks-sharded
+    tags: ["goleak"]
+    commands:
+      - func: bootstrap-mongo-orchestration
+        vars:
+          TOPOLOGY: "sharded_cluster"
+          AUTH: "noauth"
+          SSL: "nossl"
+      - func: run-goleak-test
+
   - name: test-load-balancer-noauth-nossl
     tags: ["load-balancer"]
     commands:
@@ -2567,7 +2578,7 @@ buildvariants:
     expansions:
       GO_DIST: "/opt/golang/go1.22"
     tasks:
-      - name: "test-goroutine-leaks"
+      - name: ".goleak"
 
   - matrix_name: "tests-rhel-36-with-zlib-support"
     tags: ["pullrequest"]

internal/test/goleak/goleak_test.go

@@ -22,77 +22,97 @@ import (
 
 var dbName = fmt.Sprintf("goleak-%d", time.Now().Unix())
 
-func TestMain(m *testing.M) {
-	goleak.VerifyTestMain(m)
-}
-
 // TestGoroutineLeak creates clients with various client configurations, runs
 // some operations with each one, then disconnects the client. It asserts that
 // no goroutines were leaked after the client is disconnected.
 func TestGoroutineLeak(t *testing.T) {
-	uri := "mongodb://localhost:27017"
-	if u := os.Getenv("MONGODB_URI"); u != "" {
-		uri = u
-	}
-
 	testCases := []struct {
 		desc string
 		opts *options.ClientOptions
 	}{
 		{
 			desc: "base",
-			opts: options.Client().ApplyURI(uri),
+			opts: options.Client(),
 		},
 		{
 			desc: "compressors=snappy",
-			opts: options.Client().ApplyURI(uri).SetCompressors([]string{"snappy"}),
+			opts: options.Client().SetCompressors([]string{"snappy"}),
 		},
 		{
 			desc: "compressors=zlib",
-			opts: options.Client().ApplyURI(uri).SetCompressors([]string{"zlib"}),
+			opts: options.Client().SetCompressors([]string{"zlib"}),
 		},
 		{
 			desc: "compressors=zstd",
-			opts: options.Client().ApplyURI(uri).SetCompressors([]string{"zstd"}),
+			opts: options.Client().SetCompressors([]string{"zstd"}),
 		},
 		{
 			desc: "minPoolSize=10",
-			opts: options.Client().ApplyURI(uri).SetMinPoolSize(10),
+			opts: options.Client().SetMinPoolSize(10),
 		},
 		{
 			desc: "serverMonitoringMode=poll",
-			opts: options.Client().ApplyURI(uri).SetServerMonitoringMode(options.ServerMonitoringModePoll),
+			opts: options.Client().SetServerMonitoringMode(options.ServerMonitoringModePoll),
 		},
 	}
+
 	for _, tc := range testCases {
 		// These can't be run in parallel because goleak currently can't filter
 		// out goroutines from other parallel subtests.
 		t.Run(tc.desc, func(t *testing.T) {
 			defer goleak.VerifyNone(t)
 
-			client, err := mongo.Connect(context.Background(), tc.opts)
+			base := options.Client()
+			if u := os.Getenv("MONGODB_URI"); u != "" {
+				base.ApplyURI(u)
+			}
+			client, err := mongo.Connect(context.Background(), base, tc.opts)
 			require.NoError(t, err)
 
 			defer func() {
 				err = client.Disconnect(context.Background())
 				require.NoError(t, err)
 			}()
 
-			coll := client.Database(dbName).Collection(collectionName(t))
+			db := client.Database(dbName)
 			defer func() {
-				err := coll.Drop(context.Background())
+				err := db.Drop(context.Background())
 				require.NoError(t, err)
 			}()
 
-			_, err = coll.InsertOne(context.Background(), bson.M{"x": 123})
+			coll := db.Collection(collectionName(t))
+
+			// Start a change stream to simulate a change listener workload.
+			cs, err := coll.Watch(context.Background(), mongo.Pipeline{})
 			require.NoError(t, err)
+			defer cs.Close(context.Background())
+
+			// Run some Insert and FindOne operations to simulate a writing and
+			// reading workload. Run 50 iterations to increase the probability
+			// that a goroutine leak will happen if a problem exists.
+			for i := 0; i < 50; i++ {
+				_, err = coll.InsertOne(context.Background(), bson.M{"x": 123})
+				require.NoError(t, err)
 
-			for i := 0; i < 20; i++ {
 				var res bson.D
 				err = coll.FindOne(context.Background(), bson.D{}).Decode(&res)
 				require.NoError(t, err)
-				time.Sleep(50 * time.Millisecond)
 			}
+
+			// Intentionally cause some timeouts. Ignore any errors.
+			for i := 0; i < 50; i++ {
+				ctx, cancel := context.WithTimeout(context.Background(), 10*time.Microsecond)
+				coll.FindOne(ctx, bson.D{}).Err()
+				cancel()
+			}
+
+			// Finish simulating the change listener workload. Use "Next" to
+			// fetch at least one change stream document batch and decode the
+			// first document.
+			cs.Next(context.Background())
+			var res bson.D
+			err = cs.Decode(&res)
+			require.NoError(t, err)
 		})
 	}
 }