Merge branch 'master' into ci/godriver-3659-await-min-pool-size-in-ust

Author: matthewdale

.evergreen/config.yml

@@ -198,7 +198,7 @@ functions:
       params:
         binary: bash
         env:
-          GO_BUILD_TAGS: cse
+          GO_BUILD_TAGS: "cse,mongointernal"
         include_expansions_in_env: ["TOPOLOGY", "AUTH", "SSL", "SKIP_CSOT_TESTS", "MONGODB_URI", "CRYPT_SHARED_LIB_PATH", "SKIP_CRYPT_SHARED_LIB", "RACE", "MONGO_GO_DRIVER_COMPRESSOR", "REQUIRE_API_VERSION", "LOAD_BALANCER"]
         args: [*task-runner, setup-test]
     - command: subprocess.exec

internal/integration/cursor_test.go

@@ -349,116 +349,192 @@ func parseMaxAwaitTime(mt *mtest.T, evt *event.CommandStartedEvent) int64 {
 	return got
 }
 
-func TestCursor_tailableAwaitData(t *testing.T) {
-	mt := mtest.New(t, mtest.NewOptions().CreateClient(false))
+func tadcFindFactory(ctx context.Context, mt *mtest.T, coll mongo.Collection) *mongo.Cursor {
+	mt.Helper()
 
-	cappedOpts := options.CreateCollection().SetCapped(true).
-		SetSizeInBytes(1024 * 64)
+	initCollection(mt, &coll)
+	cur, err := coll.Find(ctx, bson.D{{"__nomatch", 1}},
+		options.Find().SetBatchSize(1).SetCursorType(options.TailableAwait))
+	require.NoError(mt, err, "Find error: %v", err)
 
-	// TODO(SERVER-96344): mongos doesn't honor a failpoint's full blockTimeMS.
-	mtOpts := mtest.NewOptions().MinServerVersion("4.4").
-		Topologies(mtest.ReplicaSet, mtest.LoadBalanced, mtest.Single).
-		CollectionCreateOptions(cappedOpts)
+	return cur
+}
 
-	mt.RunOpts("apply remaining timeoutMS if less than maxAwaitTimeMS", mtOpts, func(mt *mtest.T) {
-		initCollection(mt, mt.Coll)
+func tadcAggregateFactory(ctx context.Context, mt *mtest.T, coll mongo.Collection) *mongo.Cursor {
+	mt.Helper()
 
-		// Create a 30ms failpoint for getMore.
-		mt.SetFailPoint(failpoint.FailPoint{
-			ConfigureFailPoint: "failCommand",
-			Mode: failpoint.Mode{
-				Times: 1,
-			},
-			Data: failpoint.Data{
-				FailCommands:    []string{"getMore"},
-				BlockConnection: true,
-				BlockTimeMS:     30,
-			},
-		})
+	initCollection(mt, &coll)
+	opts := options.Aggregate()
+	pipeline := mongo.Pipeline{{{"$changeStream", bson.D{{"fullDocument", "default"}}}},
+		{{"$match", bson.D{
+			{"operationType", "insert"},
+			{"fullDocment.__nomatch", 1},
+		}}},
+	}
 
-		// Create a find cursor with a 100ms maxAwaitTimeMS and a tailable awaitData
-		// cursor type.
-		opts := options.Find().
-			SetBatchSize(1).
-			SetMaxAwaitTime(100 * time.Millisecond).
-			SetCursorType(options.TailableAwait)
+	cursor, err := coll.Aggregate(ctx, pipeline, opts)
+	require.NoError(mt, err, "Aggregate error: %v", err)
 
-		cursor, err := mt.Coll.Find(context.Background(), bson.D{{"x", 2}}, opts)
-		require.NoError(mt, err)
+	return cursor
+}
 
-		defer cursor.Close(context.Background())
+func tadcRunCommandCursorFactory(ctx context.Context, mt *mtest.T, coll mongo.Collection) *mongo.Cursor {
+	mt.Helper()
 
-		// Use a 200ms timeout that caps the lifetime of cursor.Next. The underlying
-		// getMore loop should run at least two times: the first getMore will block
-		// for 30ms on the getMore and then an additional 100ms for the
-		// maxAwaitTimeMS. The second getMore will then use the remaining ~70ms
-		// left on the timeout.
-		ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
-		defer cancel()
+	initCollection(mt, &coll)
 
-		// Iterate twice to force a getMore
-		cursor.Next(ctx)
+	cur, err := coll.Database().RunCommandCursor(ctx, bson.D{
+		{"find", coll.Name()},
+		{"filter", bson.D{{"__nomatch", 1}}},
+		{"tailable", true},
+		{"awaitData", true},
+		{"batchSize", int32(1)},
+	})
+	require.NoError(mt, err, "RunCommandCursor error: %v", err)
 
-		mt.ClearEvents()
-		cursor.Next(ctx)
+	return cur
+}
 
-		require.Error(mt, cursor.Err(), "expected error from cursor.Next")
-		assert.ErrorIs(mt, cursor.Err(), context.DeadlineExceeded, "expected context deadline exceeded error")
+// For tailable awaitData cursors, the maxTimeMS for a getMore should be
+// min(maxAwaitTimeMS, remaining timeoutMS - minRoundTripTime) to allow the
+// server more opportunities to respond with an empty batch before a
+// client-side timeout.
+func TestCursor_tailableAwaitData_applyRemainingTimeout(t *testing.T) {
+	// These values reflect what is used in the unified spec tests, see
+	// DRIVERS-2868.
+	const timeoutMS = 200
+	const maxAwaitTimeMS = 100
+	const blockTimeMS = 30
+	const getMoreBound = 71
+
+	// TODO(GODRIVER-3328): mongos doesn't honor a failpoint's full blockTimeMS.
+	baseTopologies := []mtest.TopologyKind{mtest.Single, mtest.LoadBalanced, mtest.ReplicaSet}
+
+	type testCase struct {
+		name       string
+		factory    func(ctx context.Context, mt *mtest.T, coll mongo.Collection) *mongo.Cursor
+		opTimeout  bool
+		topologies []mtest.TopologyKind
+	}
 
-		// Collect all started events to find the getMore commands.
-		startedEvents := mt.GetAllStartedEvents()
+	cases := []testCase{
+		// TODO(GODRIVER-2944): "find" cursors are tested in the CSOT unified spec
+		// tests for tailable/awaitData cursors and so these tests can be removed
+		// once the driver supports timeoutMode.
+		{
+			name:       "find client-level timeout",
+			factory:    tadcFindFactory,
+			topologies: baseTopologies,
+			opTimeout:  false,
+		},
+		{
+			name:       "find operation-level timeout",
+			factory:    tadcFindFactory,
+			topologies: baseTopologies,
+			opTimeout:  true,
+		},
 
-		var getMoreStartedEvents []*event.CommandStartedEvent
-		for _, evt := range startedEvents {
-			if evt.CommandName == "getMore" {
-				getMoreStartedEvents = append(getMoreStartedEvents, evt)
-			}
-		}
+		// There is no analogue to tailable/awaiData cursor unified spec tests for
+		// aggregate and runnCommand.
+		{
+			name:       "aggregate with changeStream client-level timeout",
+			factory:    tadcAggregateFactory,
+			topologies: []mtest.TopologyKind{mtest.ReplicaSet, mtest.LoadBalanced},
+			opTimeout:  false,
+		},
+		{
+			name:       "aggregate with changeStream operation-level timeout",
+			factory:    tadcAggregateFactory,
+			topologies: []mtest.TopologyKind{mtest.ReplicaSet, mtest.LoadBalanced},
+			opTimeout:  true,
+		},
+		{
+			name:       "runCommandCursor client-level timeout",
+			factory:    tadcRunCommandCursorFactory,
+			topologies: baseTopologies,
+			opTimeout:  false,
+		},
+		{
+			name:       "runCommandCursor operation-level timeout",
+			factory:    tadcRunCommandCursorFactory,
+			topologies: baseTopologies,
+			opTimeout:  true,
+		},
+	}
 
-		// The first getMore should have a maxTimeMS of <= 100ms.
-		assert.LessOrEqual(mt, parseMaxAwaitTime(mt, getMoreStartedEvents[0]), int64(100))
+	mt := mtest.New(t, mtest.NewOptions().CreateClient(false).MinServerVersion("4.2"))
 
-		// The second getMore should have a maxTimeMS of <=71, indicating that we
-		// are using the time remaining in the context rather than the
-		// maxAwaitTimeMS.
-		assert.LessOrEqual(mt, parseMaxAwaitTime(mt, getMoreStartedEvents[1]), int64(71))
-	})
+	for _, tc := range cases {
+		// Reset the collection between test cases to avoid leaking timeouts
+		// between tests.
+		cappedOpts := options.CreateCollection().SetCapped(true).SetSizeInBytes(1024 * 64)
+		caseOpts := mtest.NewOptions().
+			CollectionCreateOptions(cappedOpts).
+			Topologies(tc.topologies...).
+			CreateClient(true)
 
-	mtOpts.Topologies(mtest.ReplicaSet, mtest.Sharded, mtest.LoadBalanced, mtest.Single)
+		if !tc.opTimeout {
+			caseOpts = caseOpts.ClientOptions(options.Client().SetTimeout(timeoutMS * time.Millisecond))
+		}
 
-	mt.RunOpts("apply maxAwaitTimeMS if less than remaining timeout", mtOpts, func(mt *mtest.T) {
-		initCollection(mt, mt.Coll)
-		mt.ClearEvents()
+		mt.RunOpts(tc.name, caseOpts, func(mt *mtest.T) {
+			mt.SetFailPoint(failpoint.FailPoint{
+				ConfigureFailPoint: "failCommand",
+				Mode:               failpoint.Mode{Times: 1},
+				Data: failpoint.Data{
+					FailCommands:    []string{"getMore"},
+					BlockConnection: true,
+					BlockTimeMS:     int32(blockTimeMS),
+				},
+			})
+
+			ctx := context.Background()
 
-		// Create a find cursor
-		opts := options.Find().SetBatchSize(1).SetMaxAwaitTime(50 * time.Millisecond)
+			var cancel context.CancelFunc
+			if tc.opTimeout {
+				ctx, cancel = context.WithTimeout(ctx, timeoutMS*time.Millisecond)
+				defer cancel()
+			}
 
-		cursor, err := mt.Coll.Find(context.Background(), bson.D{}, opts)
-		require.NoError(mt, err)
+			cur := tc.factory(ctx, mt, *mt.Coll)
+			defer func() { assert.NoError(mt, cur.Close(context.Background())) }()
 
-		_ = mt.GetStartedEvent() // Empty find from started list.
+			require.NoError(mt, cur.Err())
 
-		defer cursor.Close(context.Background())
+			cur.SetMaxAwaitTime(maxAwaitTimeMS * time.Millisecond)
 
-		ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
-		defer cancel()
+			mt.ClearEvents()
 
-		// Iterate twice to force a getMore
-		cursor.Next(ctx)
-		cursor.Next(ctx)
+			assert.False(mt, cur.Next(ctx))
 
-		cmd := mt.GetStartedEvent().Command
+			require.Error(mt, cur.Err(), "expected error from cursor.Next")
+			assert.ErrorIs(mt, cur.Err(), context.DeadlineExceeded, "expected context deadline exceeded error")
 
-		maxTimeMSRaw, err := cmd.LookupErr("maxTimeMS")
-		require.NoError(mt, err)
+			getMoreEvts := []*event.CommandStartedEvent{}
+			for _, evt := range mt.GetAllStartedEvents() {
+				if evt.CommandName == "getMore" {
+					getMoreEvts = append(getMoreEvts, evt)
+				}
+			}
 
-		got, ok := maxTimeMSRaw.AsInt64OK()
-		require.True(mt, ok)
+			// It's possible that three getMore events are called: 100ms, 70ms, and
+			// then some small leftover of remaining time (e.g. 20µs).
+			require.GreaterOrEqual(mt, len(getMoreEvts), 2)
 
-		assert.LessOrEqual(mt, got, int64(50))
-	})
+			// The first getMore should have a maxTimeMS of <= 100ms but greater
+			// than 71ms, indicating that the maxAwaitTimeMS was used.
+			assert.LessOrEqual(mt, parseMaxAwaitTime(mt, getMoreEvts[0]), int64(maxAwaitTimeMS))
+			assert.Greater(mt, parseMaxAwaitTime(mt, getMoreEvts[0]), int64(getMoreBound))
+
+			// The second getMore should have a maxTimeMS of <=71, indicating that we
+			// are using the time remaining in the context rather than the
+			// maxAwaitTimeMS.
+			assert.LessOrEqual(mt, parseMaxAwaitTime(mt, getMoreEvts[1]), int64(getMoreBound))
+		})
+	}
 }
 
+// For tailable awaitData cursors, the maxTimeMS for a getMore should be
 func TestCursor_tailableAwaitData_ShortCircuitingGetMore(t *testing.T) {
 	mt := mtest.New(t, mtest.NewOptions().CreateClient(false))
 

internal/integration/mongointernal_test.go

@@ -0,0 +1,98 @@
+// Copyright (C) MongoDB, Inc. 2025-present.
+//
+// Licensed under the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License. You may obtain
+// a copy of the License at http://www.apache.org/licenses/LICENSE-2.0
+
+//go:build mongointernal
+
+package integration
+
+import (
+	"context"
+	"testing"
+
+	"go.mongodb.org/mongo-driver/v2/bson"
+	"go.mongodb.org/mongo-driver/v2/internal/assert"
+	"go.mongodb.org/mongo-driver/v2/internal/integration/mtest"
+	"go.mongodb.org/mongo-driver/v2/internal/require"
+	"go.mongodb.org/mongo-driver/v2/mongo"
+	"go.mongodb.org/mongo-driver/v2/x/bsonx/bsoncore"
+)
+
+func TestNewSessionWithLSID(t *testing.T) {
+	mt := mtest.New(t)
+
+	mt.Run("can be used to pass a specific session ID to CRUD commands", func(mt *mtest.T) {
+		mt.Parallel()
+
+		// Create a session ID document, which is a BSON document with field
+		// "id" containing a 16-byte UUID (binary subtype 4).
+		sessionID := bson.Raw(bsoncore.NewDocumentBuilder().
+			AppendBinary("id", 4, []byte{0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15}).
+			Build())
+
+		sess := mongo.NewSessionWithLSID(mt.Client, sessionID)
+
+		ctx := mongo.NewSessionContext(context.Background(), sess)
+		_, err := mt.Coll.InsertOne(ctx, bson.D{{"foo", "bar"}})
+		require.NoError(mt, err)
+
+		evt := mt.GetStartedEvent()
+		val, err := evt.Command.LookupErr("lsid")
+		require.NoError(mt, err, "lsid should be present in the command document")
+
+		doc, ok := val.DocumentOK()
+		require.True(mt, ok, "lsid should be a document")
+
+		assert.Equal(mt, sessionID, doc)
+	})
+
+	mt.Run("EndSession panics", func(mt *mtest.T) {
+		mt.Parallel()
+
+		sessionID := bson.Raw(bsoncore.NewDocumentBuilder().
+			AppendBinary("id", 4, []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}).
+			Build())
+		sess := mongo.NewSessionWithLSID(mt.Client, sessionID)
+
+		// Use a defer-recover block to catch the expected panic and assert that
+		// the recovered error is not nil.
+		defer func() {
+			err := recover()
+			assert.NotNil(mt, err, "expected EndSession to panic")
+		}()
+
+		// Expect this call to panic.
+		sess.EndSession(context.Background())
+
+		// We expect that calling EndSession on a Session returned by
+		// NewSessionWithLSID panics. This code will only be reached if EndSession
+		// doesn't panic.
+		t.Errorf("expected EndSession to panic")
+	})
+
+	mt.Run("ClientSession.SetServer panics", func(mt *mtest.T) {
+		mt.Parallel()
+
+		sessionID := bson.Raw(bsoncore.NewDocumentBuilder().
+			AppendBinary("id", 4, []byte{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}).
+			Build())
+		sess := mongo.NewSessionWithLSID(mt.Client, sessionID)
+
+		// Use a defer-recover block to catch the expected panic and assert that
+		// the recovered error is not nil.
+		defer func() {
+			err := recover()
+			assert.NotNil(mt, err, "expected ClientSession.SetServer to panic")
+		}()
+
+		// Expect this call to panic.
+		sess.ClientSession().SetServer()
+
+		// We expect that calling ClientSession.SetServer on a Session returned
+		// by NewSessionWithLSID panics. This code will only be reached if
+		// ClientSession.SetServer doesn't panic.
+		t.Errorf("expected ClientSession.SetServer to panic")
+	})
+}