GODRIVER-3181 Read server responses in the background after op timeout.

Author: matthewdale

internal/integration/csot_test.go

@@ -208,15 +208,15 @@ func (t *T) cleanup() {
 // Run creates a new T instance for a sub-test and runs the given callback. It also creates a new collection using the
 // given name which is available to the callback through the T.Coll variable and is dropped after the callback
 // returns.
-func (t *T) Run(name string, callback func(*T)) {
+func (t *T) Run(name string, callback func(mt *T)) {
 	t.RunOpts(name, NewOptions(), callback)
 }
 
 // RunOpts creates a new T instance for a sub-test with the given options. If the current environment does not satisfy
 // constraints specified in the options, the new sub-test will be skipped automatically. If the test is not skipped,
 // the callback will be run with the new T instance. RunOpts creates a new collection with the given name which is
 // available to the callback through the T.Coll variable and is dropped after the callback returns.
-func (t *T) RunOpts(name string, opts *Options, callback func(*T)) {
+func (t *T) RunOpts(name string, opts *Options, callback func(mt *T)) {
 	t.T.Run(name, func(wrapped *testing.T) {
 		sub := newT(wrapped, t.baseOpts, opts)
 

internal/integration/mtest/mongotest.go

@@ -61,6 +61,18 @@ var (
 		"unpin when a new transaction is started":                     "Implement GODRIVER-3034",
 		"unpin when a non-transaction write operation uses a session": "Implement GODRIVER-3034",
 		"unpin when a non-transaction read operation uses a session":  "Implement GODRIVER-3034",
+
+		// DRIVERS-2722: Setting "maxTimeMS" on a command that creates a cursor
+		// also limits the lifetime of the cursor. That may be surprising to
+		// users, so omit "maxTimeMS" from operations that return user-managed
+		// cursors.
+		"timeoutMS can be overridden for a find":                                               "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"timeoutMS can be configured for an operation - find on collection":                    "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"timeoutMS can be configured for an operation - aggregate on collection":               "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"timeoutMS can be configured for an operation - aggregate on database":                 "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"operation is retried multiple times for non-zero timeoutMS - find on collection":      "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"operation is retried multiple times for non-zero timeoutMS - aggregate on collection": "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
+		"operation is retried multiple times for non-zero timeoutMS - aggregate on database":   "maxTimeMS is disabled on find and aggregate. See DRIVERS-2722.",
 	}
 
 	logMessageValidatorTimeout = 10 * time.Millisecond

internal/integration/unified/unified_spec_runner.go

@@ -948,7 +948,12 @@ func aggregate(a aggregateParams, opts ...options.Lister[options.AggregateOption
 		Crypt(a.client.cryptFLE).
 		ServerAPI(a.client.serverAPI).
 		HasOutputStage(hasOutputStage).
-		Timeout(a.client.timeout)
+		Timeout(a.client.timeout).
+		// Omit "maxTimeMS" from operations that return a user-managed cursor to
+		// prevent confusing "cursor not found" errors.
+		//
+		// See DRIVERS-2722 for more detail.
+		OmitMaxTimeMS(true)
 
 	if args.AllowDiskUse != nil {
 		op.AllowDiskUse(*args.AllowDiskUse)
@@ -1292,11 +1297,20 @@ func (coll *Collection) Find(ctx context.Context, filter interface{},
 	if err != nil {
 		return nil, err
 	}
-	return coll.find(ctx, filter, args)
+
+	// Omit "maxTimeMS" from operations that return a user-managed cursor to
+	// prevent confusing "cursor not found" errors.
+	//
+	// See DRIVERS-2722 for more detail.
+	return coll.find(ctx, filter, true, args)
 }
 
-func (coll *Collection) find(ctx context.Context, filter interface{},
-	args *options.FindOptions) (cur *Cursor, err error) {
+func (coll *Collection) find(
+	ctx context.Context,
+	filter interface{},
+	omitMaxTimeMS bool,
+	args *options.FindOptions,
+) (cur *Cursor, err error) {
 
 	if ctx == nil {
 		ctx = context.Background()
@@ -1334,7 +1348,7 @@ func (coll *Collection) find(ctx context.Context, filter interface{},
 		CommandMonitor(coll.client.monitor).ServerSelector(selector).
 		ClusterClock(coll.client.clock).Database(coll.db.name).Collection(coll.name).
 		Deployment(coll.client.deployment).Crypt(coll.client.cryptFLE).ServerAPI(coll.client.serverAPI).
-		Timeout(coll.client.timeout).Logger(coll.client.logger)
+		Timeout(coll.client.timeout).Logger(coll.client.logger).OmitMaxTimeMS(omitMaxTimeMS)
 
 	cursorOpts := coll.client.createBaseCursorOptions()
 
@@ -1499,7 +1513,7 @@ func (coll *Collection) FindOne(ctx context.Context, filter interface{},
 	if err != nil {
 		return nil
 	}
-	cursor, err := coll.find(ctx, filter, newFindArgsFromFindOneArgs(args))
+	cursor, err := coll.find(ctx, filter, false, newFindArgsFromFindOneArgs(args))
 	return &SingleResult{
 		ctx:      ctx,
 		cur:      cursor,

mongo/collection.go

@@ -442,6 +442,12 @@ func (bc *BatchCursor) getMore(ctx context.Context) {
 		Crypt:          bc.crypt,
 		ServerAPI:      bc.serverAPI,
 
+		// Omit the automatically-calculated maxTimeMS because setting maxTimeMS
+		// on a non-awaitData cursor causes a server error. For awaitData
+		// cursors, maxTimeMS is set by the above CommandFn when maxAwaitTime is
+		// specified.
+		OmitMaxTimeMS: true,
+
 		// No read preference is passed to the getMore command,
 		// resulting in the default read preference: "primaryPreferred".
 		// Since this could be confusing, and there is no requirement

x/mongo/driver/batch_cursor.go

@@ -509,14 +509,28 @@ func ExtractErrorFromServerResponse(doc bsoncore.Document) error {
 			errmsg = "command failed"
 		}
 
-		return Error{
+		err := Error{
 			Code:            code,
 			Message:         errmsg,
 			Name:            codeName,
 			Labels:          labels,
 			TopologyVersion: tv,
 			Raw:             doc,
 		}
+
+		// If we get a MaxTimeMSExpired error, assume that the error was caused
+		// by setting "maxTimeMS" on the command based on the context deadline
+		// or on "timeoutMS". In that case, make the error wrap
+		// context.DeadlineExceeded so that users can always check
+		//
+		//  errors.Is(err, context.DeadlineExceeded)
+		//
+		// for either client-side or server-side timeouts.
+		if err.Code == 50 {
+			err.Wrapped = context.DeadlineExceeded
+		}
+
+		return err
 	}
 
 	if len(wcError.WriteErrors) > 0 || wcError.WriteConcernError != nil {

x/mongo/driver/errors.go

@@ -1186,7 +1186,7 @@ func (op Operation) addBatchArray(dst []byte) []byte {
 
 func (op Operation) createLegacyHandshakeWireMessage(
 	ctx context.Context,
-	maxTimeMS uint64,
+	maxTimeMS int64,
 	dst []byte,
 	desc description.SelectedServer,
 ) ([]byte, startedInformation, error) {
@@ -1245,7 +1245,7 @@ func (op Operation) createLegacyHandshakeWireMessage(
 	// If maxTimeMS is greater than 0 append it to wire message. A maxTimeMS value of 0 only explicitly
 	// specifies the default behavior of no timeout server-side.
 	if maxTimeMS > 0 {
-		dst = bsoncore.AppendInt64Element(dst, "maxTimeMS", int64(maxTimeMS))
+		dst = bsoncore.AppendInt64Element(dst, "maxTimeMS", maxTimeMS)
 	}
 
 	dst, _ = bsoncore.AppendDocumentEnd(dst, idx)
@@ -1266,7 +1266,7 @@ func (op Operation) createLegacyHandshakeWireMessage(
 
 func (op Operation) createMsgWireMessage(
 	ctx context.Context,
-	maxTimeMS uint64,
+	maxTimeMS int64,
 	dst []byte,
 	desc description.SelectedServer,
 	conn *mnet.Connection,
@@ -1316,7 +1316,7 @@ func (op Operation) createMsgWireMessage(
 	// If maxTimeMS is greater than 0 append it to wire message. A maxTimeMS value of 0 only explicitly
 	// specifies the default behavior of no timeout server-side.
 	if maxTimeMS > 0 {
-		dst = bsoncore.AppendInt64Element(dst, "maxTimeMS", int64(maxTimeMS))
+		dst = bsoncore.AppendInt64Element(dst, "maxTimeMS", maxTimeMS)
 	}
 
 	dst = bsoncore.AppendStringElement(dst, "$db", op.Database)
@@ -1362,7 +1362,7 @@ func isLegacyHandshake(op Operation, desc description.SelectedServer) bool {
 
 func (op Operation) createWireMessage(
 	ctx context.Context,
-	maxTimeMS uint64,
+	maxTimeMS int64,
 	dst []byte,
 	desc description.SelectedServer,
 	conn *mnet.Connection,
@@ -1620,7 +1620,7 @@ func (op Operation) addClusterTime(dst []byte, desc description.SelectedServer)
 // if the ctx is a Timeout context. If the context is not a Timeout context, it uses the
 // operation's MaxTimeMS if set. If no MaxTimeMS is set on the operation, and context is
 // not a Timeout context, calculateMaxTimeMS returns 0.
-func (op Operation) calculateMaxTimeMS(ctx context.Context, rttMin time.Duration, rttStats string) (uint64, error) {
+func (op Operation) calculateMaxTimeMS(ctx context.Context, rttMin time.Duration, rttStats string) (int64, error) {
 	if op.OmitMaxTimeMS {
 		return 0, nil
 	}
@@ -1637,13 +1637,23 @@ func (op Operation) calculateMaxTimeMS(ctx context.Context, rttMin time.Duration
 	maxTimeMS := int64((remainingTimeout - rttMin + time.Millisecond - 1) / time.Millisecond)
 	if maxTimeMS <= 0 {
 		return 0, fmt.Errorf(
-			"remaining time %v until context deadline is less than or equal to rtt minimum: %w\n%v",
+			"remaining time %v until context deadline is less than or equal to min network round-trip time %v (%v): %w",
 			remainingTimeout,
-			ErrDeadlineWouldBeExceeded,
-			rttStats)
+			rttMin,
+			rttStats,
+			ErrDeadlineWouldBeExceeded)
 	}
 
-	return uint64(maxTimeMS), nil
+	// The server will return a "BadValue" error if maxTimeMS is greater
+	// than the maximum positive int32 value (about 24.9 days). If the
+	// user specified a timeout value greater than that,  omit maxTimeMS
+	// and let the client-side timeout handle cancelling the op if the
+	// timeout is ever reached.
+	if maxTimeMS > math.MaxInt32 {
+		return 0, nil
+	}
+
+	return maxTimeMS, nil
 }
 
 // updateClusterTimes updates the cluster times for the session and cluster clock attached to this

x/mongo/driver/operation.go

@@ -48,6 +48,7 @@ type Aggregate struct {
 	hasOutputStage           bool
 	customOptions            map[string]bsoncore.Value
 	timeout                  *time.Duration
+	omitMaxTimeMS            bool
 
 	result driver.CursorResponse
 }
@@ -110,6 +111,7 @@ func (a *Aggregate) Execute(ctx context.Context) error {
 		IsOutputAggregate:              a.hasOutputStage,
 		Timeout:                        a.timeout,
 		Name:                           driverutil.AggregateOp,
+		OmitMaxTimeMS:                  a.omitMaxTimeMS,
 	}.Execute(ctx)
 
 }
@@ -404,3 +406,14 @@ func (a *Aggregate) Timeout(timeout *time.Duration) *Aggregate {
 	a.timeout = timeout
 	return a
 }
+
+// OmitMaxTimeMS omits the automatically-calculated "maxTimeMS" from the
+// command.
+func (a *Aggregate) OmitMaxTimeMS(omit bool) *Aggregate {
+	if a == nil {
+		a = new(Aggregate)
+	}
+
+	a.omitMaxTimeMS = omit
+	return a
+}

x/mongo/driver/operation/aggregate.go

@@ -61,6 +61,7 @@ type Find struct {
 	serverAPI           *driver.ServerAPIOptions
 	timeout             *time.Duration
 	logger              *logger.Logger
+	omitMaxTimeMS       bool
 }
 
 // NewFind constructs and returns a new Find.
@@ -107,6 +108,7 @@ func (f *Find) Execute(ctx context.Context) error {
 		Timeout:           f.timeout,
 		Logger:            f.logger,
 		Name:              driverutil.FindOp,
+		OmitMaxTimeMS:     f.omitMaxTimeMS,
 	}.Execute(ctx)
 }
 
@@ -547,3 +549,14 @@ func (f *Find) Logger(logger *logger.Logger) *Find {
 	f.logger = logger
 	return f
 }
+
+// OmitMaxTimeMS omits the automatically-calculated "maxTimeMS" from the
+// command.
+func (f *Find) OmitMaxTimeMS(omit bool) *Find {
+	if f == nil {
+		f = new(Find)
+	}
+
+	f.omitMaxTimeMS = omit
+	return f
+}

x/mongo/driver/operation/find.go

@@ -286,7 +286,7 @@ func TestOperation(t *testing.T) {
 			rtt      RTTMonitor
 			rttMin   time.Duration
 			rttStats string
-			want     uint64
+			want     int64
 			err      error
 		}{
 			{
@@ -644,6 +644,35 @@ func TestOperation(t *testing.T) {
 		// the TransientTransactionError label.
 		assert.Equal(t, err, context.Canceled, "expected context.Canceled error, got %v", err)
 	})
+	t.Run("ErrDeadlineWouldBeExceeded wraps context.DeadlineExceeded", func(t *testing.T) {
+		// Create a deployment that returns a server that reports a 90th
+		// percentile RTT of 1 minute.
+		d := new(mockDeployment)
+		d.returns.server = mockServer{
+			conn:       mnet.NewConnection(&mockConnection{}),
+			rttMonitor: mockRTTMonitor{min: 1 * time.Minute},
+		}
+
+		// Create an operation with a Timeout specified to enable CSOT behavior.
+		var dur time.Duration
+		op := Operation{
+			Database:   "foobar",
+			Deployment: d,
+			CommandFn: func(dst []byte, desc description.SelectedServer) ([]byte, error) {
+				return dst, nil
+			},
+			Timeout: &dur,
+		}
+
+		// Call the operation with a context with a deadline less than the 90th
+		// percentile RTT configured above.
+		ctx, cancel := context.WithTimeout(context.Background(), 1*time.Second)
+		defer cancel()
+		err := op.Execute(ctx)
+
+		assert.ErrorIs(t, err, ErrDeadlineWouldBeExceeded)
+		assert.ErrorIs(t, err, context.DeadlineExceeded)
+	})
 }
 
 func createExhaustServerResponse(response bsoncore.Document, moreToCome bool) []byte {
@@ -709,6 +738,25 @@ func (m *mockServerSelector) String() string {
 	panic("not implemented")
 }
 
+type mockServer struct {
+	conn       *mnet.Connection
+	err        error
+	rttMonitor RTTMonitor
+}
+
+func (ms mockServer) Connection(context.Context) (*mnet.Connection, error) { return ms.conn, ms.err }
+func (ms mockServer) RTTMonitor() RTTMonitor                               { return ms.rttMonitor }
+
+type mockRTTMonitor struct {
+	ewma  time.Duration
+	min   time.Duration
+	stats string
+}
+
+func (mrm mockRTTMonitor) EWMA() time.Duration { return mrm.ewma }
+func (mrm mockRTTMonitor) Min() time.Duration  { return mrm.min }
+func (mrm mockRTTMonitor) Stats() string       { return mrm.stats }
+
 type mockConnection struct {
 	// parameters
 	pWriteWM []byte

x/mongo/driver/operation_test.go

@@ -76,6 +76,10 @@ type connection struct {
 
 	driverConnectionID int64
 	generation         uint64
+
+	// awaitingResponse indicates that the server response was not completely
+	// read before returning the connection to the pool.
+	awaitingResponse bool
 }
 
 // newConnection handles the creation of a connection. It does not connect the connection.
@@ -370,8 +374,16 @@ func (c *connection) readWireMessage(ctx context.Context) ([]byte, error) {
 
 	dst, errMsg, err := c.read(ctx)
 	if err != nil {
-		// We closeConnection the connection because we don't know if there are other bytes left to read.
-		c.close()
+		if nerr := net.Error(nil); errors.As(err, &nerr) && nerr.Timeout() {
+			// If the error was a timeout error, instead of closing the
+			// connection mark it as awaiting response so the pool can read the
+			// response before making it available to other operations.
+			c.awaitingResponse = true
+		} else {
+			// Otherwise, and close the connection because we don't know what
+			// the connection state is.
+			c.close()
+		}
 		message := errMsg
 		if errors.Is(err, io.EOF) {
 			message = "socket was unexpectedly closed"