Use mongoutil.TimeoutWithinContext over validChangeStreamTimeouts for change streams

Author: prestonvasquez

mongo/change_stream.go

@@ -12,7 +12,6 @@ import (
 	"fmt"
 	"reflect"
 	"strconv"
-	"time"
 
 	"go.mongodb.org/mongo-driver/v2/bson"
 	"go.mongodb.org/mongo-driver/v2/internal/csot"
@@ -103,33 +102,6 @@ type changeStreamConfig struct {
 	crypt          driver.Crypt
 }
 
-// validChangeStreamTimeouts will return "false" if maxAwaitTimeMS is set,
-// timeoutMS is set to a non-zero value, and maxAwaitTimeMS is greater than or
-// equal to timeoutMS. Otherwise, the timeouts are valid.
-func validChangeStreamTimeouts(ctx context.Context, cs *ChangeStream) bool {
-	if cs.options == nil || cs.client == nil {
-		return true
-	}
-
-	maxAwaitTime := cs.options.MaxAwaitTime
-	timeout := cs.client.timeout
-
-	if maxAwaitTime == nil {
-		return true
-	}
-
-	if deadline, ok := ctx.Deadline(); ok {
-		ctxTimeout := time.Until(deadline)
-		timeout = &ctxTimeout
-	}
-
-	if timeout == nil {
-		return true
-	}
-
-	return *timeout <= 0 || *maxAwaitTime < *timeout
-}
-
 func newChangeStream(ctx context.Context, config changeStreamConfig, pipeline interface{},
 	opts ...options.Lister[options.ChangeStreamOptions]) (*ChangeStream, error) {
 	if ctx == nil {
@@ -696,10 +668,33 @@ func (cs *ChangeStream) next(ctx context.Context, nonBlocking bool) bool {
 }
 
 func (cs *ChangeStream) loopNext(ctx context.Context, nonBlocking bool) {
-	if !validChangeStreamTimeouts(ctx, cs) {
-		cs.err = fmt.Errorf("MaxAwaitTime must be less than the operation timeout")
+	// To avoid unnecessary socket timeouts, we attempt to short-circuit tailable
+	// awaitData "getMore" operations by ensuring that the maxAwaitTimeMS is less
+	// than the operation timeout.
+	//
+	// The specifications assume that drivers iteratively apply the timeout
+	// provided at the constructor level (e.g., (*collection).Find) for tailable
+	// awaitData cursors:
+	//
+	//	If set, drivers MUST apply the timeoutMS option to the initial aggregate
+	//	operation. Drivers MUST also apply the original timeoutMS value to each
+	//	next call on the change stream but MUST NOT use it to derive a maxTimeMS
+	//	field for getMore commands.
+	//
+	// The Go Driver might decide to support the above behavior with DRIVERS-2722.
+	// The principal concern is that it would be unexpected for users to apply an
+	// operation-level timeout via contexts to a constructor and then that timeout
+	// later be applied while working with a resulting cursor. Instead, it is more
+	// idiomatic to apply the timeout to the context passed to Next or TryNext.
+	if cs.options != nil && !nonBlocking {
+		maxAwaitTime := cs.cursorOptions.MaxAwaitTime
+
+		// If maxAwaitTime is not set, this check is unnecessary.
+		if maxAwaitTime != nil && !mongoutil.TimeoutWithinContext(ctx, *maxAwaitTime) {
+			cs.err = fmt.Errorf("MaxAwaitTime must be less than the operation timeout")
 
-		return
+			return
+		}
 	}
 
 	// Apply the client-level timeout if the operation-level timeout is not set.

mongo/change_stream_test.go

@@ -7,12 +7,9 @@
 package mongo
 
 import (
-	"context"
 	"testing"
-	"time"
 
 	"go.mongodb.org/mongo-driver/v2/internal/assert"
-	"go.mongodb.org/mongo-driver/v2/mongo/options"
 )
 
 func TestChangeStream(t *testing.T) {
@@ -30,96 +27,3 @@ func TestChangeStream(t *testing.T) {
 		assert.Nil(t, err, "Close error: %v", err)
 	})
 }
-
-func TestValidChangeStreamTimeouts(t *testing.T) {
-	t.Parallel()
-
-	newDurPtr := func(dur time.Duration) *time.Duration {
-		return &dur
-	}
-
-	tests := []struct {
-		name                     string
-		parent                   context.Context
-		maxAwaitTimeout, timeout *time.Duration
-		wantTimeout              time.Duration
-		want                     bool
-	}{
-		{
-			name:            "no context deadline and no timeouts",
-			parent:          context.Background(),
-			maxAwaitTimeout: nil,
-			timeout:         nil,
-			wantTimeout:     0,
-			want:            true,
-		},
-		{
-			name:            "no context deadline and maxAwaitTimeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: newDurPtr(1),
-			timeout:         nil,
-			wantTimeout:     0,
-			want:            true,
-		},
-		{
-			name:            "no context deadline and timeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: nil,
-			timeout:         newDurPtr(1),
-			wantTimeout:     0,
-			want:            true,
-		},
-		{
-			name:            "no context deadline and maxAwaitTime gt timeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: newDurPtr(2),
-			timeout:         newDurPtr(1),
-			wantTimeout:     0,
-			want:            false,
-		},
-		{
-			name:            "no context deadline and maxAwaitTime lt timeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: newDurPtr(1),
-			timeout:         newDurPtr(2),
-			wantTimeout:     0,
-			want:            true,
-		},
-		{
-			name:            "no context deadline and maxAwaitTime eq timeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: newDurPtr(1),
-			timeout:         newDurPtr(1),
-			wantTimeout:     0,
-			want:            false,
-		},
-		{
-			name:            "no context deadline and maxAwaitTime with negative timeout",
-			parent:          context.Background(),
-			maxAwaitTimeout: newDurPtr(1),
-			timeout:         newDurPtr(-1),
-			wantTimeout:     0,
-			want:            true,
-		},
-	}
-
-	for _, test := range tests {
-		test := test // Capture the range variable
-
-		t.Run(test.name, func(t *testing.T) {
-			t.Parallel()
-
-			cs := &ChangeStream{
-				options: &options.ChangeStreamOptions{
-					MaxAwaitTime: test.maxAwaitTimeout,
-				},
-				client: &Client{
-					timeout: test.timeout,
-				},
-			}
-
-			got := validChangeStreamTimeouts(test.parent, cs)
-			assert.Equal(t, test.want, got)
-		})
-	}
-}