Use connectTimeoutMS for heartbeatTimeout

GODRIVER-726

Change-Id: I7041da51a9a056a149e1efc8a0baf8187e840d1a

Author: skriptble

x/mongo/driver/topology/server.go

@@ -370,6 +370,7 @@ func (s *Server) heartbeat(conn connection.Connection) (description.Server, conn
 			opts := []connection.Option{
 				connection.WithConnectTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
 				connection.WithReadTimeout(func(time.Duration) time.Duration { return s.cfg.heartbeatTimeout }),
+				connection.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 an empty

x/mongo/driver/topology/server_options.go

@@ -32,7 +32,7 @@ type serverConfig struct {
 func newServerConfig(opts ...ServerOption) (*serverConfig, error) {
 	cfg := &serverConfig{
 		heartbeatInterval: 10 * time.Second,
-		heartbeatTimeout:  30 * time.Second,
+		heartbeatTimeout:  10 * time.Second,
 		maxConns:          100,
 		maxIdleConns:      100,
 		registry:          defaultRegistry,

x/mongo/driver/topology/topology_options.go

@@ -70,6 +70,7 @@ func WithConnString(fn func(connstring.ConnString) connstring.ConnString) Option
 		c.seedList = cs.Hosts
 
 		if cs.ConnectTimeout > 0 {
+			c.serverOpts = append(c.serverOpts, WithHeartbeatTimeout(func(time.Duration) time.Duration { return cs.ConnectTimeout }))
 			connOpts = append(connOpts, connection.WithConnectTimeout(func(time.Duration) time.Duration { return cs.ConnectTimeout }))
 		}
 

x/network/integration/main_test.go

@@ -160,3 +160,25 @@ func (d *dialer) lenclosed() int {
 	defer d.Unlock()
 	return len(d.closed)
 }
+
+// bootstrapConnections lists for num connections on the returned address. The user provided run
+// function will be called with the accepted connection. The user is responsible for closing the
+// connection.
+func bootstrapConnections(t *testing.T, num int, run func(net.Conn)) net.Addr {
+	l, err := net.Listen("tcp", "localhost:0")
+	if err != nil {
+		t.Errorf("Could not set up a listener: %v", err)
+		t.FailNow()
+	}
+	go func() {
+		for i := 0; i < num; i++ {
+			c, err := l.Accept()
+			if err != nil {
+				t.Errorf("Could not accept a connection: %v", err)
+			}
+			go run(c)
+		}
+		_ = l.Close()
+	}()
+	return l.Addr()
+}

x/network/integration/topology_test.go

@@ -10,6 +10,7 @@ import (
 	"context"
 	"net"
 	"testing"
+	"time"
 
 	"github.com/mongodb/mongo-go-driver/x/mongo/driver/topology"
 	"github.com/mongodb/mongo-go-driver/x/network/connection"
@@ -85,6 +86,40 @@ func TestTopologyTopology(t *testing.T) {
 					d.lenclosed(), d.lenopened())
 			}
 		})
+		t.Run("can disconnect from hung server", func(t *testing.T) {
+			bctx, cancel := context.WithCancel(context.Background())
+			defer cancel()
+			start := make(chan struct{}, 1)
+			addr := bootstrapConnections(t, 5, func(_ net.Conn) {
+				start <- struct{}{}
+				<-bctx.Done()
+			})
+			topo, err := topology.New(
+				topology.WithConnString(
+					func(connstring.ConnString) connstring.ConnString {
+						cs, err := connstring.Parse("mongodb://" + addr.String() + "/")
+						noerr(t, err)
+						cs.ConnectTimeout = 20 * time.Millisecond
+						return cs
+					},
+				),
+			)
+			noerr(t, err)
+			err = topo.Connect(context.TODO())
+			noerr(t, err)
+			<-start
+			done := make(chan struct{}, 1)
+			go func() {
+				err := topo.Disconnect(bctx)
+				noerr(t, err)
+				done <- struct{}{}
+			}()
+			select {
+			case <-done:
+			case <-time.After(300 * time.Millisecond):
+				t.Error("Could not disconnect a topology within required timelimit")
+			}
+		})
 	})
 	t.Run("Connect", func(t *testing.T) {
 		t.Run("can reconnect a disconnected topology", func(t *testing.T) {