GODRIVER-3255 Await heartbeat checks upto freq when polling

mongo/integration/sdam_prose_test.go

@@ -11,6 +11,7 @@ import (
 	"net"
 	"os"
 	"runtime"
+	"sync/atomic"
 	"testing"
 	"time"
 
@@ -144,7 +145,7 @@ func TestSDAMProse(t *testing.T) {
 		})
 	})
 
-	mt.RunOpts("client waits between failed Hellos", mtest.NewOptions().MinServerVersion("4.9").Topologies(mtest.Single), func(mt *mtest.T) {
+	mt.RunOpts("client waits between failed Hellos", mtest.NewOptions().MinServerVersion("4.9"), func(mt *mtest.T) {
 		// Force hello requests to fail 5 times.
 		mt.SetFailPoint(mtest.FailPoint{
 			ConfigureFailPoint: "failCommand",
@@ -232,4 +233,36 @@ func TestServerHeartbeatStartedEvent(t *testing.T) {
 		}
 		assert.Equal(t, expectedEvents, actualEvents)
 	})
+
+	mt := mtest.New(t)
+
+	mt.Run("polling must await frequency", func(mt *mtest.T) {
+		var heartbeatStartedCount atomic.Int64
+		servers := map[string]bool{}
+
+		serverMonitor := &event.ServerMonitor{
+			ServerHeartbeatStarted: func(*event.ServerHeartbeatStartedEvent) {
+				heartbeatStartedCount.Add(1)
+			},
+			TopologyDescriptionChanged: func(evt *event.TopologyDescriptionChangedEvent) {
+				for _, srv := range evt.NewDescription.Servers {
+					servers[srv.Addr.String()] = true
+				}
+			},
+		}
+
+		// Create a client with  heartbeatFrequency=100ms,
+		// serverMonitoringMode=poll. Use SDAM to record the number of times the
+		// a heartbeat is started and the number of servers discovered.
+		mt.ResetClient(options.Client().
+			SetServerMonitor(serverMonitor).
+			SetServerMonitoringMode(options.ServerMonitoringModePoll))
+
+		// Per specifications, minHeartbeatFrequencyMS=500ms. So, within the first
+		// 500ms the heartbeatStartedCount should be LEQ to the number of discovered
+		// servers.
+		time.Sleep(500 * time.Millisecond)
+
+		assert.LessOrEqual(mt, heartbeatStartedCount.Load(), int64(len(servers)))
+	})
 }

x/mongo/driver/topology/server.go

@@ -666,7 +666,7 @@ func (s *Server) update() {
 			s.monitorOnce.Do(s.rttMonitor.connect)
 		}
 
-		if isStreamable(s) || connectionIsStreaming || transitionedFromNetworkError {
+		if isStreamingEnabled(s) && (isStreamable(s) || connectionIsStreaming) || transitionedFromNetworkError {
 			continue
 		}