GODRIVER-1695 always use heartbeatTimeout for heartbeat connection (#470)

Author: Isabella Siu

x/mongo/driver/topology/server.go

@@ -438,6 +438,26 @@ func (s *Server) updateDescription(desc description.Server) {
 	s.subLock.Unlock()
 }
 
+// createConnection creates a new connection instance but does not call connect on it. The caller must call connect
+// before the connection can be used for network operations.
+func (s *Server) createConnection(ctx context.Context) (*connection, error) {
+	opts := []ConnectionOption{
+		WithConnectTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
+		WithReadTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
+		WithWriteTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
+		// We override whatever handshaker is currently attached to the options with a basic
+		// one because need to make sure we don't do auth.
+		WithHandshaker(func(h Handshaker) Handshaker {
+			return operation.NewIsMaster().AppName(s.cfg.appname).Compressors(s.cfg.compressionOpts)
+		}),
+		// Override any command monitors specified in options with nil to avoid monitoring heartbeats.
+		WithMonitor(func(*event.CommandMonitor) *event.CommandMonitor { return nil }),
+	}
+	opts = append(s.cfg.connectionOpts, opts...)
+
+	return newConnection(ctx, s.address, opts...)
+}
+
 // heartbeat sends a heartbeat to the server using the given connection. The connection can be nil.
 func (s *Server) heartbeat(conn *connection) (description.Server, *connection) {
 	const maxRetry = 2
@@ -467,25 +487,7 @@ func (s *Server) heartbeat(conn *connection) (description.Server, *connection) {
 		}
 
 		if conn == nil {
-			opts := []ConnectionOption{
-				WithConnectTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
-				WithReadTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
-				WithWriteTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
-			}
-			opts = append(opts, s.cfg.connectionOpts...)
-			// We override whatever handshaker is currently attached to the options with a basic
-			// one because need to make sure we don't do auth.
-			opts = append(opts, WithHandshaker(func(h Handshaker) Handshaker {
-				now = time.Now()
-				return operation.NewIsMaster().AppName(s.cfg.appname).Compressors(s.cfg.compressionOpts)
-			}))
-
-			// Override any command monitors specified in options with nil to avoid monitoring heartbeats.
-			opts = append(opts, WithMonitor(func(*event.CommandMonitor) *event.CommandMonitor {
-				return nil
-			}))
-
-			conn, err = newConnection(ctx, s.address, opts...)
+			conn, err = s.createConnection(ctx)
 
 			conn.connect(ctx)
 

x/mongo/driver/topology/server_test.go

@@ -17,6 +17,7 @@ import (
 
 	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/internal/testutil/assert"
 	"go.mongodb.org/mongo-driver/x/bsonx/bsoncore"
 	"go.mongodb.org/mongo-driver/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/x/mongo/driver/address"
@@ -297,6 +298,28 @@ func TestServer(t *testing.T) {
 		require.Nil(t, err, "error from NewServer: %v", err)
 		require.Equal(t, name, s.cfg.appname, "expected appname to be: %v, got: %v", name, s.cfg.appname)
 	})
+	t.Run("createConnection overwrites WithSocketTimeout", func(t *testing.T) {
+		socketTimeout := 40 * time.Second
+
+		s, err := NewServer(
+			address.Address("localhost"),
+			WithConnectionOptions(func(connOpts ...ConnectionOption) []ConnectionOption {
+				return append(
+					connOpts,
+					WithReadTimeout(func(time.Duration) time.Duration { return socketTimeout }),
+					WithWriteTimeout(func(time.Duration) time.Duration { return socketTimeout }),
+				)
+			}),
+		)
+		assert.Nil(t, err, "NewServer error: %v", err)
+
+		conn, err := s.createConnection(context.Background())
+		assert.Nil(t, err, "createConnection error: %v", err)
+
+		assert.Equal(t, s.cfg.heartbeatTimeout, 10*time.Second, "expected heartbeatTimeout to be: %v, got: %v", 10*time.Second, s.cfg.heartbeatTimeout)
+		assert.Equal(t, s.cfg.heartbeatTimeout, conn.readTimeout, "expected readTimeout to be: %v, got: %v", s.cfg.heartbeatTimeout, conn.readTimeout)
+		assert.Equal(t, s.cfg.heartbeatTimeout, conn.writeTimeout, "expected writeTimeout to be: %v, got: %v", s.cfg.heartbeatTimeout, conn.writeTimeout)
+	})
 }
 
 func includesMetadata(t *testing.T, wm []byte) bool {