GODRIVER-1442 Fix connection error handling in topology.Server (#259)

Author: Divjot Arora

mongo/integration/primary_stepdown_test.go

@@ -124,4 +124,24 @@ func TestConnectionsSurvivePrimaryStepDown(t *testing.T) {
 			})
 		}
 	})
+	mt.RunOpts("network errors", mtest.NewOptions().ClientOptions(clientOpts).MinServerVersion("4.0"), func(mt *mtest.T) {
+		// expect that a server's connection pool will be cleared if a non-timeout network error occurs during an
+		// operation
+
+		clearPoolChan()
+		mt.SetFailPoint(mtest.FailPoint{
+			ConfigureFailPoint: "failCommand",
+			Mode: mtest.FailPointMode{
+				Times: 1,
+			},
+			Data: mtest.FailPointData{
+				FailCommands:    []string{"insert"},
+				CloseConnection: true,
+			},
+		})
+
+		_, err := mt.Coll.InsertOne(mtest.Background, bson.D{{"test", 1}})
+		assert.NotNil(mt, err, "expected InsertOne error, got nil")
+		assert.True(mt, isPoolCleared(), "expected pool to be cleared but was not")
+	})
 }

x/mongo/driver/errors.go

@@ -190,6 +190,7 @@ type Error struct {
 	Message string
 	Labels  []string
 	Name    string
+	Wrapped error
 }
 
 // UnsupportedStorageEngine returns whether e came as a result of an unsupported storage engine

x/mongo/driver/operation.go

@@ -546,7 +546,7 @@ func (op Operation) roundTrip(ctx context.Context, conn Connection, wm []byte) (
 		if op.Client != nil && op.Client.Committing {
 			labels = append(labels, UnknownTransactionCommitResult)
 		}
-		return nil, Error{Message: err.Error(), Labels: labels}
+		return nil, Error{Message: err.Error(), Labels: labels, Wrapped: err}
 	}
 
 	wm, err = conn.ReadWireMessage(ctx, wm[:0])
@@ -561,7 +561,7 @@ func (op Operation) roundTrip(ctx context.Context, conn Connection, wm []byte) (
 		if op.Client != nil && op.Client.Committing {
 			labels = append(labels, UnknownTransactionCommitResult)
 		}
-		return nil, Error{Message: err.Error(), Labels: labels}
+		return nil, Error{Message: err.Error(), Labels: labels, Wrapped: err}
 	}
 
 	// decompress wiremessage
@@ -588,7 +588,7 @@ func (op *Operation) moreToComeRoundTrip(ctx context.Context, conn Connection, w
 		if op.Client != nil {
 			op.Client.MarkDirty()
 		}
-		err = Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}}
+		err = Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}, Wrapped: err}
 	}
 	return bsoncore.BuildDocument(nil, bsoncore.AppendInt32Element(nil, "ok", 1)), err
 }

x/mongo/driver/topology/server.go

@@ -243,16 +243,16 @@ func (s *Server) Connection(ctx context.Context) (driver.Connection, error) {
 	conn, err := s.pool.get(ctx)
 	if err != nil {
 		s.sem.Release(1)
-		connErr, ok := err.(ConnectionError)
-		if !ok {
+		wrappedConnErr := unwrapConnectionError(err)
+		if wrappedConnErr == nil {
 			return nil, err
 		}
 
 		// Since the only kind of ConnectionError we receive from pool.Get will be an initialization
 		// error, we should set the description.Server appropriately.
 		desc := description.Server{
 			Kind:      description.Unknown,
-			LastError: connErr.Wrapped,
+			LastError: wrappedConnErr,
 		}
 		s.updateDescription(desc, false)
 
@@ -317,8 +317,9 @@ func (s *Server) RequestImmediateCheck() {
 
 // ProcessError handles SDAM error handling and implements driver.ErrorProcessor.
 func (s *Server) ProcessError(err error) {
-	// Invalidate server description if not master or node recovering error occurs
-	if cerr, ok := err.(driver.Error); ok && (cerr.NetworkError() || cerr.NodeIsRecovering() || cerr.NotMaster()) {
+	// Invalidate server description if not master or node recovering error occurs.
+	// These errors can be reported as a command error or a write concern error.
+	if cerr, ok := err.(driver.Error); ok && (cerr.NodeIsRecovering() || cerr.NotMaster()) {
 		desc := s.Description()
 		desc.Kind = description.Unknown
 		desc.LastError = err
@@ -345,15 +346,16 @@ func (s *Server) ProcessError(err error) {
 		return
 	}
 
-	ne, ok := err.(ConnectionError)
-	if !ok {
+	wrappedConnErr := unwrapConnectionError(err)
+	if wrappedConnErr == nil {
 		return
 	}
 
-	if netErr, ok := ne.Wrapped.(net.Error); ok && netErr.Timeout() {
+	// Ignore transient timeout errors.
+	if netErr, ok := wrappedConnErr.(net.Error); ok && netErr.Timeout() {
 		return
 	}
-	if ne.Wrapped == context.Canceled || ne.Wrapped == context.DeadlineExceeded {
+	if wrappedConnErr == context.Canceled || wrappedConnErr == context.DeadlineExceeded {
 		return
 	}
 
@@ -362,6 +364,7 @@ func (s *Server) ProcessError(err error) {
 	desc.LastError = err
 	// updates description to unknown
 	s.updateDescription(desc, false)
+	s.pool.clear()
 }
 
 // update handles performing heartbeats and updating any subscribers of the
@@ -551,7 +554,7 @@ func (s *Server) heartbeat(conn *connection) (description.Server, *connection) {
 		if err != nil {
 			saved = err
 			conn = nil
-			if _, ok := err.(ConnectionError); ok {
+			if wrappedConnErr := unwrapConnectionError(err); wrappedConnErr != nil {
 				s.pool.drain()
 				// If the server is not connected, give up and exit loop
 				if s.Description().Kind == description.Unknown {
@@ -637,3 +640,18 @@ func (ss *ServerSubscription) Unsubscribe() error {
 
 	return nil
 }
+
+// unwrapConnectionError returns the connection error wrapped by err, or nil if err does not wrap a connection error.
+func unwrapConnectionError(err error) error {
+	connErr, ok := err.(ConnectionError)
+	if ok {
+		return connErr.Wrapped
+	}
+
+	driverErr, ok := err.(driver.Error)
+	if ok && driverErr.NetworkError() {
+		return driverErr.Wrapped
+	}
+
+	return nil
+}