GODRIVER-1578 Set server to Unknown for set name mismatches in direct connections (#368)

Author: Divjot Arora

data/server-discovery-and-monitoring/single/direct_connection_wrong_set_name.json

@@ -0,0 +1,63 @@
+{
+  "description": "Direct connection to RSPrimary with wrong set name",
+  "uri": "mongodb://a/?directConnection=true&replicaSet=rs",
+  "phases": [
+    {
+      "responses": [
+        [
+          "a:27017",
+          {
+            "ok": 1,
+            "ismaster": true,
+            "hosts": [
+              "a:27017",
+              "b:27017"
+            ],
+            "setName": "wrong",
+            "minWireVersion": 0,
+            "maxWireVersion": 6
+          }
+        ]
+      ],
+      "outcome": {
+        "servers": {
+          "a:27017": {
+            "type": "Unknown"
+          }
+        },
+        "topologyType": "Single",
+        "logicalSessionTimeoutMinutes": null,
+        "setName": "rs"
+      }
+    },
+    {
+      "responses": [
+        [
+          "a:27017",
+          {
+            "ok": 1,
+            "ismaster": true,
+            "hosts": [
+              "a:27017",
+              "b:27017"
+            ],
+            "setName": "rs",
+            "minWireVersion": 0,
+            "maxWireVersion": 6
+          }
+        ]
+      ],
+      "outcome": {
+        "servers": {
+          "a:27017": {
+            "type": "RSPrimary",
+            "setName": "rs"
+          }
+        },
+        "topologyType": "Single",
+        "logicalSessionTimeoutMinutes": null,
+        "setName": "rs"
+      }
+    }
+  ]
+}

data/server-discovery-and-monitoring/single/direct_connection_wrong_set_name.yml

@@ -0,0 +1,38 @@
+description: Direct connection to RSPrimary with wrong set name
+uri: mongodb://a/?directConnection=true&replicaSet=rs
+phases:
+- responses:
+  - - a:27017
+    - ok: 1
+      ismaster: true
+      hosts:
+      - a:27017
+      - b:27017
+      setName: wrong
+      minWireVersion: 0
+      maxWireVersion: 6
+  outcome:
+    servers:
+      a:27017:
+        type: Unknown
+    topologyType: Single
+    logicalSessionTimeoutMinutes: 
+    setName: rs
+- responses:
+  - - a:27017
+    - ok: 1
+      ismaster: true
+      hosts:
+      - a:27017
+      - b:27017
+      setName: rs
+      minWireVersion: 0
+      maxWireVersion: 6
+  outcome:
+    servers:
+      a:27017:
+        type: RSPrimary
+        setName: rs
+    topologyType: Single
+    logicalSessionTimeoutMinutes: 
+    setName: rs

x/mongo/driver/topology/fsm.go

@@ -29,9 +29,13 @@ func newFSM() *fsm {
 	return new(fsm)
 }
 
-// apply should operate on immutable TopologyDescriptions and Descriptions. This way we don't have to
-// lock for the entire time we're applying server description.
-func (f *fsm) apply(s description.Server) (description.Topology, error) {
+// apply takes a new server description and modifies the FSM's topology description based on it. It returns the
+// updated topology description as well as a server description. The returned server description is either the same
+// one that was passed in, or a new one in the case that it had to be changed.
+//
+// apply should operation on immutable descriptions so we don't have to lock for the entire time we're applying the
+// server description.
+func (f *fsm) apply(s description.Server) (description.Topology, description.Server, error) {
 
 	newServers := make([]description.Server, len(f.Servers))
 	copy(newServers, f.Servers)
@@ -62,12 +66,12 @@ func (f *fsm) apply(s description.Server) (description.Topology, error) {
 	}
 
 	if _, ok := f.findServer(s.Addr); !ok {
-		return f.Topology, nil
+		return f.Topology, s, nil
 	}
 
 	if s.WireVersion != nil {
 		if s.WireVersion.Max < supportedWireVersions.Min {
-			return description.Topology{}, fmt.Errorf(
+			return description.Topology{}, s, fmt.Errorf(
 				"server at %s reports wire version %d, but this version of the Go driver requires "+
 					"at least %d (MongoDB %s)",
 				s.Addr.String(),
@@ -78,7 +82,7 @@ func (f *fsm) apply(s description.Server) (description.Topology, error) {
 		}
 
 		if s.WireVersion.Min > supportedWireVersions.Max {
-			return description.Topology{}, fmt.Errorf(
+			return description.Topology{}, s, fmt.Errorf(
 				"server at %s requires wire version %d, but this version of the Go driver only "+
 					"supports up to %d",
 				s.Addr.String(),
@@ -88,23 +92,24 @@ func (f *fsm) apply(s description.Server) (description.Topology, error) {
 		}
 	}
 
+	updatedDesc := s
 	switch f.Kind {
 	case description.Unknown:
-		f.applyToUnknown(s)
+		updatedDesc = f.applyToUnknown(s)
 	case description.Sharded:
-		f.applyToSharded(s)
+		updatedDesc = f.applyToSharded(s)
 	case description.ReplicaSetNoPrimary:
-		f.applyToReplicaSetNoPrimary(s)
+		updatedDesc = f.applyToReplicaSetNoPrimary(s)
 	case description.ReplicaSetWithPrimary:
-		f.applyToReplicaSetWithPrimary(s)
+		updatedDesc = f.applyToReplicaSetWithPrimary(s)
 	case description.Single:
-		f.applyToSingle(s)
+		updatedDesc = f.applyToSingle(s)
 	}
 
-	return f.Topology, nil
+	return f.Topology, updatedDesc, nil
 }
 
-func (f *fsm) applyToReplicaSetNoPrimary(s description.Server) {
+func (f *fsm) applyToReplicaSetNoPrimary(s description.Server) description.Server {
 	switch s.Kind {
 	case description.Standalone, description.Mongos:
 		f.removeServerByAddr(s.Addr)
@@ -115,9 +120,11 @@ func (f *fsm) applyToReplicaSetNoPrimary(s description.Server) {
 	case description.Unknown, description.RSGhost:
 		f.replaceServer(s)
 	}
+
+	return s
 }
 
-func (f *fsm) applyToReplicaSetWithPrimary(s description.Server) {
+func (f *fsm) applyToReplicaSetWithPrimary(s description.Server) description.Server {
 	switch s.Kind {
 	case description.Standalone, description.Mongos:
 		f.removeServerByAddr(s.Addr)
@@ -130,39 +137,53 @@ func (f *fsm) applyToReplicaSetWithPrimary(s description.Server) {
 		f.replaceServer(s)
 		f.checkIfHasPrimary()
 	}
+
+	return s
 }
 
-func (f *fsm) applyToSharded(s description.Server) {
+func (f *fsm) applyToSharded(s description.Server) description.Server {
 	switch s.Kind {
 	case description.Mongos, description.Unknown:
 		f.replaceServer(s)
 	case description.Standalone, description.RSPrimary, description.RSSecondary, description.RSArbiter, description.RSMember, description.RSGhost:
 		f.removeServerByAddr(s.Addr)
 	}
+
+	return s
 }
 
-func (f *fsm) applyToSingle(s description.Server) {
+func (f *fsm) applyToSingle(s description.Server) description.Server {
 	switch s.Kind {
 	case description.Unknown:
 		f.replaceServer(s)
 	case description.Standalone, description.Mongos:
 		if f.SetName != "" {
 			f.removeServerByAddr(s.Addr)
-			return
+			return s
 		}
 
 		f.replaceServer(s)
 	case description.RSPrimary, description.RSSecondary, description.RSArbiter, description.RSMember, description.RSGhost:
+		// A replica set name can be provided when creating a direct connection. In this case, if the set name returned
+		// by the isMaster response doesn't match up with the one provided during configuration, the server description
+		// is replaced with a default Unknown description.
+		//
+		// We create a new server description rather than doing s.Kind = description.Unknown because the other fields,
+		// such as RTT, need to be cleared for Unknown descriptions as well.
 		if f.SetName != "" && f.SetName != s.SetName {
-			f.removeServerByAddr(s.Addr)
-			return
+			s = description.Server{
+				Addr: s.Addr,
+				Kind: description.Unknown,
+			}
 		}
 
 		f.replaceServer(s)
 	}
+
+	return s
 }
 
-func (f *fsm) applyToUnknown(s description.Server) {
+func (f *fsm) applyToUnknown(s description.Server) description.Server {
 	switch s.Kind {
 	case description.Mongos:
 		f.setKind(description.Sharded)
@@ -177,6 +198,8 @@ func (f *fsm) applyToUnknown(s description.Server) {
 	case description.Unknown, description.RSGhost:
 		f.replaceServer(s)
 	}
+
+	return s
 }
 
 func (f *fsm) checkIfHasPrimary() {

x/mongo/driver/topology/sdam_spec_test.go

@@ -119,6 +119,13 @@ func setUpFSM(t *testing.T, uri string) *fsm {
 		fsm.Kind = description.Single
 	}
 
+	// GODRIVER-1578: The new test requires support for directConnection, which will be added in follow-up ticket
+	// GODRIVER-1486. In this case, look for "directconnection" in the parsed URI's set of unrecognized options and
+	// emulate direct connection behavior.
+	if dcValues, ok := cs.UnknownOptions["directconnection"]; ok && dcValues[0] == "true" {
+		fsm.Kind = description.Single
+	}
+
 	for _, host := range cs.Hosts {
 		fsm.Servers = append(fsm.Servers, description.Server{Addr: address.Address(host).Canonicalize()})
 	}
@@ -133,7 +140,7 @@ func applyResponses(f *fsm, responses []response) error {
 			return err
 		}
 		server := description.NewServer(address.Address(response.Host), bsoncore.Document(doc))
-		_, err = f.apply(server)
+		_, _, err = f.apply(server)
 
 		if err != nil {
 			return err

x/mongo/driver/topology/server.go

@@ -107,9 +107,14 @@ type Server struct {
 	subscriptionsClosed bool
 }
 
+// updateTopologyCallback is a callback used to create a server that should be called when the parent Topology instance
+// should be updated based on a new server description. The callback must return the server description that should be
+// stored by the server.
+type updateTopologyCallback func(description.Server) description.Server
+
 // ConnectServer creates a new Server and then initializes it using the
 // Connect method.
-func ConnectServer(addr address.Address, updateCallback func(description.Server), opts ...ServerOption) (*Server, error) {
+func ConnectServer(addr address.Address, updateCallback updateTopologyCallback, opts ...ServerOption) (*Server, error) {
 	srvr, err := NewServer(addr, opts...)
 	if err != nil {
 		return nil, err
@@ -166,7 +171,7 @@ func NewServer(addr address.Address, opts ...ServerOption) (*Server, error) {
 
 // Connect initializes the Server by starting background monitoring goroutines.
 // This method must be called before a Server can be used.
-func (s *Server) Connect(updateCallback func(description.Server)) error {
+func (s *Server) Connect(updateCallback updateTopologyCallback) error {
 	if !atomic.CompareAndSwapInt32(&s.connectionstate, disconnected, connected) {
 		return ErrServerConnected
 	}
@@ -191,7 +196,7 @@ func (s *Server) Disconnect(ctx context.Context) error {
 		return ErrServerClosed
 	}
 
-	s.updateTopologyCallback.Store((func(description.Server))(nil))
+	s.updateTopologyCallback.Store((updateTopologyCallback)(nil))
 
 	// For every call to Connect there must be at least 1 goroutine that is
 	// waiting on the done channel.
@@ -446,12 +451,13 @@ func (s *Server) updateDescription(desc description.Server, initial bool) {
 		//  ¯\_(ツ)_/¯
 		_ = recover()
 	}()
-	s.desc.Store(desc)
 
-	callback, ok := s.updateTopologyCallback.Load().(func(description.Server))
+	// Use the updateTopologyCallback to update the parent Topology and get the description that should be stored.
+	callback, ok := s.updateTopologyCallback.Load().(updateTopologyCallback)
 	if ok && callback != nil {
-		callback(desc)
+		desc = callback(desc)
 	}
+	s.desc.Store(desc)
 
 	s.subLock.Lock()
 	for _, c := range s.subscribers {

x/mongo/driver/topology/server_test.go

@@ -243,7 +243,12 @@ func TestServer(t *testing.T) {
 	t.Run("update topology", func(t *testing.T) {
 		var updated atomic.Value // bool
 		updated.Store(false)
-		s, err := ConnectServer(address.Address("localhost"), func(description.Server) { updated.Store(true) })
+
+		updateCallback := func(desc description.Server) description.Server {
+			updated.Store(true)
+			return desc
+		}
+		s, err := ConnectServer(address.Address("localhost"), updateCallback)
 		require.NoError(t, err)
 		s.updateDescription(description.Server{Addr: s.address}, false)
 		require.True(t, updated.Load().(bool))