GODRIVER-2065 Don't send the operation if the Context deadline is before (now + minimum RTT). (#761)

Author: matthewdale

internal/testutil/assert/assert.go

@@ -81,8 +81,7 @@ func NotNil(t testing.TB, obj interface{}, msg string, args ...interface{}) {
 	}
 }
 
-// Soon runs the provided callback for a maximum of timeoutMS milliseconds. It returns the callback error
-// if the callback returned and ErrCallbackTimedOut if the timeout expired.
+// Soon runs the provided callback for a maximum of timeoutMS milliseconds.
 func Soon(t testing.TB, callback func(), timeout time.Duration) {
 	t.Helper()
 

mongo/change_stream_deployment.go

@@ -8,6 +8,7 @@ package mongo
 
 import (
 	"context"
+	"time"
 
 	"go.mongodb.org/mongo-driver/mongo/description"
 	"go.mongodb.org/mongo-driver/x/mongo/driver"
@@ -35,6 +36,10 @@ func (c *changeStreamDeployment) Connection(context.Context) (driver.Connection,
 	return c.conn, nil
 }
 
+func (c *changeStreamDeployment) MinRTT() time.Duration {
+	return c.server.MinRTT()
+}
+
 func (c *changeStreamDeployment) ProcessError(err error, conn driver.Connection) driver.ProcessErrorResult {
 	ep, ok := c.server.(driver.ErrorProcessor)
 	if !ok {

mongo/integration/client_test.go

@@ -9,6 +9,7 @@ package integration
 import (
 	"context"
 	"fmt"
+	"net"
 	"os"
 	"reflect"
 	"strings"
@@ -53,6 +54,46 @@ func (e *negateCodec) DecodeValue(ectx bsoncodec.DecodeContext, vr bsonrw.ValueR
 	return nil
 }
 
+var _ options.ContextDialer = &slowConnDialer{}
+
+// A slowConnDialer dials connections that delay network round trips by the given delay duration.
+type slowConnDialer struct {
+	dialer *net.Dialer
+	delay  time.Duration
+}
+
+func newSlowConnDialer(delay time.Duration) *slowConnDialer {
+	return &slowConnDialer{
+		dialer: &net.Dialer{},
+		delay:  delay,
+	}
+}
+
+func (scd *slowConnDialer) DialContext(ctx context.Context, network, address string) (net.Conn, error) {
+	conn, err := scd.dialer.DialContext(ctx, network, address)
+	if err != nil {
+		return nil, err
+	}
+	return &slowConn{
+		Conn:  conn,
+		delay: scd.delay,
+	}, nil
+}
+
+var _ net.Conn = &slowConn{}
+
+// slowConn is a net.Conn that delays all calls to Read() by given delay durations. All other
+// net.Conn functions behave identically to the embedded net.Conn.
+type slowConn struct {
+	net.Conn
+	delay time.Duration
+}
+
+func (sc *slowConn) Read(b []byte) (n int, err error) {
+	time.Sleep(sc.delay)
+	return sc.Conn.Read(b)
+}
+
 func TestClient(t *testing.T) {
 	mt := mtest.New(t, noClientOpts)
 	defer mt.Close()
@@ -467,6 +508,95 @@ func TestClient(t *testing.T) {
 		err := mt.Client.Ping(context.Background(), readpref.Primary())
 		assert.Nil(t, err, "unexpected error calling Ping: %v", err)
 	})
+
+	mt.Run("minimum RTT is monitored", func(mt *mtest.T) {
+		if testing.Short() {
+			t.Skip("skipping integration test in short mode")
+		}
+
+		// Reset the client with a dialer that delays all network round trips by 300ms and set the
+		// heartbeat interval to 100ms to reduce the time it takes to collect RTT samples.
+		mt.ResetClient(options.Client().
+			SetDialer(newSlowConnDialer(300 * time.Millisecond)).
+			SetHeartbeatInterval(100 * time.Millisecond))
+
+		// Assert that the minimum RTT is eventually >250ms.
+		topo := getTopologyFromClient(mt.Client)
+		assert.Soon(mt, func() {
+			for {
+				time.Sleep(100 * time.Millisecond)
+
+				// Wait for all of the server's minimum RTTs to be >250ms.
+				done := true
+				for _, desc := range topo.Description().Servers {
+					server, err := topo.FindServer(desc)
+					assert.Nil(mt, err, "FindServer error: %v", err)
+					if server.MinRTT() <= 250*time.Millisecond {
+						done = false
+					}
+				}
+				if done {
+					return
+				}
+			}
+		}, 10*time.Second)
+	})
+
+	// Test that if the minimum RTT is greater than the remaining timeout for an operation, the
+	// operation is not sent to the server and no connections are closed.
+	mt.Run("minimum RTT used to prevent sending requests", func(mt *mtest.T) {
+		if testing.Short() {
+			t.Skip("skipping integration test in short mode")
+		}
+
+		// Assert that we can call Ping with a 250ms timeout.
+		ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond)
+		defer cancel()
+		err := mt.Client.Ping(ctx, nil)
+		assert.Nil(mt, err, "Ping error: %v", err)
+
+		// Reset the client with a dialer that delays all network round trips by 300ms and set the
+		// heartbeat interval to 100ms to reduce the time it takes to collect RTT samples.
+		tpm := newTestPoolMonitor()
+		mt.ResetClient(options.Client().
+			SetPoolMonitor(tpm.PoolMonitor).
+			SetDialer(newSlowConnDialer(300 * time.Millisecond)).
+			SetHeartbeatInterval(100 * time.Millisecond))
+
+		// Assert that the minimum RTT is eventually >250ms.
+		topo := getTopologyFromClient(mt.Client)
+		assert.Soon(mt, func() {
+			for {
+				time.Sleep(100 * time.Millisecond)
+
+				// Wait for all of the server's minimum RTTs to be >250ms.
+				done := true
+				for _, desc := range topo.Description().Servers {
+					server, err := topo.FindServer(desc)
+					assert.Nil(mt, err, "FindServer error: %v", err)
+					if server.MinRTT() <= 250*time.Millisecond {
+						done = false
+					}
+				}
+				if done {
+					return
+				}
+			}
+		}, 10*time.Second)
+
+		// Once we've waited for the minimum RTT for the single server to be >250ms, run a bunch of
+		// Ping operations with a timeout of 250ms and expect that they return errors.
+		for i := 0; i < 10; i++ {
+			ctx, cancel = context.WithTimeout(context.Background(), 250*time.Millisecond)
+			err := mt.Client.Ping(ctx, nil)
+			cancel()
+			assert.NotNil(mt, err, "expected Ping to return an error")
+		}
+
+		// Assert that the Ping timeouts result in no connections being closed.
+		closed := len(tpm.Events(func(e *event.PoolEvent) bool { return e.Type == event.ConnectionClosed }))
+		assert.Equal(t, 0, closed, "expected no connections to be closed")
+	})
 }
 
 func TestClientStress(t *testing.T) {

mongo/integration/mtest/opmsg_deployment.go

@@ -8,6 +8,7 @@ package mtest
 
 import (
 	"context"
+	"time"
 
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
@@ -133,6 +134,11 @@ func (md *mockDeployment) Connection(context.Context) (driver.Connection, error)
 	return md.conn, nil
 }
 
+// MinRTT always returns 0. It implements the driver.Server interface.
+func (md *mockDeployment) MinRTT() time.Duration {
+	return 0
+}
+
 // Connect is a no-op method which implements the driver.Connector interface.
 func (md *mockDeployment) Connect() error {
 	return nil

mongo/integration/mtest/proxy_dialer.go

@@ -131,10 +131,8 @@ func (p *proxyDialer) Messages() []*ProxyMessage {
 	p.Lock()
 	defer p.Unlock()
 
-	copiedMessages := make([]*ProxyMessage, 0, len(p.messages))
-	for _, msg := range p.messages {
-		copiedMessages = append(copiedMessages, msg)
-	}
+	copiedMessages := make([]*ProxyMessage, len(p.messages))
+	copy(copiedMessages, p.messages)
 	return copiedMessages
 }
 

mongo/integration/sessions_test.go

@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"go.mongodb.org/mongo-driver/bson"
+	"go.mongodb.org/mongo-driver/internal"
 	"go.mongodb.org/mongo-driver/internal/testutil/assert"
 	"go.mongodb.org/mongo-driver/mongo"
 	"go.mongodb.org/mongo-driver/mongo/integration/mtest"
@@ -130,22 +131,33 @@ func TestSessions(t *testing.T) {
 	mt.RunOpts("cluster time is updated from handshakes", clusterTimeHandshakeMtOpts, func(mt *mtest.T) {
 		err := mt.Client.Ping(mtest.Background, mtest.PrimaryRp)
 		assert.Nil(mt, err, "Ping error: %v", err)
-		msgPairs := mt.GetProxiedMessages()
-		assert.True(mt, len(msgPairs) > 2, "expected more than two messages, got %d", len(msgPairs))
-
-		for idx, pair := range mt.GetProxiedMessages() {
-			// Get the $clusterTime value sent to the server. The first three messages are the handshakes for the
-			// heartbeat, RTT, and application connections. These should not contain $clusterTime because they happen on
-			// connections that don't know the server's wire version and therefore don't know if the server supports
-			// $clusterTime.
-			_, err = pair.Sent.Command.LookupErr("$clusterTime")
-			if idx <= 2 {
-				assert.NotNil(mt, err, "expected no $clusterTime field in command %s", pair.Sent.Command)
-				continue
-			}
 
-			// All messages after the first two should contain $clusterTime.
-			assert.Nil(mt, err, "expected $clusterTime field in command %s", pair.Sent.Command)
+		// Assert that all sent commands (including handshake commands) include a "$clusterTime" in
+		// the command document.
+		for _, pair := range mt.GetProxiedMessages() {
+			_, err := pair.Sent.Command.LookupErr("$clusterTime")
+			hasClusterTime := err == nil
+
+			switch pair.CommandName {
+			// If the command is either legacy hello or "hello" (used as the first message in any
+			// handshake or the checks from the heartbeat or RTT monitors), expect that there is no
+			// "$clusterTime" because the connection doesn't know the server's wire version yet so
+			// it doesn't know if the connection supports "$clusterTime".
+			case internal.LegacyHello, "hello":
+				assert.False(
+					mt,
+					hasClusterTime,
+					"expected no $clusterTime field in command %s",
+					pair.Sent.Command)
+			// If the command is anything else (including other handshake commands), assert that the
+			// command includes "$clusterTime".
+			default:
+				assert.True(
+					mt,
+					hasClusterTime,
+					"expected $clusterTime field in in Ping command %s",
+					pair.Sent.Command)
+			}
 		}
 	})
 

x/mongo/driver/batch_cursor.go

@@ -5,6 +5,7 @@ import (
 	"errors"
 	"fmt"
 	"strings"
+	"time"
 
 	"go.mongodb.org/mongo-driver/bson/bsontype"
 	"go.mongodb.org/mongo-driver/event"
@@ -426,6 +427,11 @@ func (lbcd *loadBalancedCursorDeployment) Connection(_ context.Context) (Connect
 	return lbcd.conn, nil
 }
 
+// MinRTT always returns 0. It implements the driver.Server interface.
+func (lbcd *loadBalancedCursorDeployment) MinRTT() time.Duration {
+	return 0
+}
+
 func (lbcd *loadBalancedCursorDeployment) ProcessError(err error, conn Connection) ProcessErrorResult {
 	return lbcd.errorProcessor.ProcessError(err, conn)
 }

x/mongo/driver/driver.go

@@ -2,6 +2,7 @@ package driver // import "go.mongodb.org/mongo-driver/x/mongo/driver"
 
 import (
 	"context"
+	"time"
 
 	"go.mongodb.org/mongo-driver/mongo/address"
 	"go.mongodb.org/mongo-driver/mongo/description"
@@ -43,6 +44,9 @@ type Subscriber interface {
 // retrieving and returning of connections.
 type Server interface {
 	Connection(context.Context) (Connection, error)
+
+	// MinRTT returns the minimum round-trip time to the server observed over the window period.
+	MinRTT() time.Duration
 }
 
 // Connection represents a connection to a MongoDB server.
@@ -194,6 +198,11 @@ func (ssd SingleConnectionDeployment) Connection(context.Context) (Connection, e
 	return ssd.C, nil
 }
 
+// MinRTT always returns 0. It implements the driver.Server interface.
+func (ssd SingleConnectionDeployment) MinRTT() time.Duration {
+	return 0
+}
+
 // TODO(GODRIVER-617): We can likely use 1 type for both the Type and the RetryMode by using
 // 2 bits for the mode and 1 bit for the type. Although in the practical sense, we might not want to
 // do that since the type of retryability is tied to the operation itself and isn't going change,

x/mongo/driver/drivertest/channel_netconn.go

@@ -76,11 +76,12 @@ func (c *ChannelNetConn) SetWriteDeadline(_ time.Time) error {
 
 // GetWrittenMessage gets the last wire message written to the connection
 func (c *ChannelNetConn) GetWrittenMessage() []byte {
-	var wm []byte
 	select {
-	case wm = <-c.Written:
+	case wm := <-c.Written:
+		return wm
+	default:
+		return nil
 	}
-	return wm
 }
 
 // AddResponse adds a response to the connection.