GODRIVER-1785 Add a status return to the ProcessError function (#547)

Author: Divjot Arora

mongo/change_stream_deployment.go

@@ -35,11 +35,11 @@ func (c *changeStreamDeployment) Connection(context.Context) (driver.Connection,
 	return c.conn, nil
 }
 
-func (c *changeStreamDeployment) ProcessError(err error, conn driver.Connection) {
+func (c *changeStreamDeployment) ProcessError(err error, conn driver.Connection) driver.ProcessErrorResult {
 	ep, ok := c.server.(driver.ErrorProcessor)
 	if !ok {
-		return
+		return driver.NoChange
 	}
 
-	ep.ProcessError(err, conn)
+	return ep.ProcessError(err, conn)
 }

x/mongo/driver/driver.go

@@ -88,11 +88,32 @@ type Compressor interface {
 	CompressWireMessage(src, dst []byte) ([]byte, error)
 }
 
+// ProcessErrorResult represents the result of a ErrorProcessor.ProcessError() call. Exact values for this type can be
+// checked directly (e.g. res == ServerMarkedUnknown), but it is recommended that applications use the ServerChanged()
+// function instead.
+type ProcessErrorResult int
+
+const (
+	// NoChange indicates that the error did not affect the state of the server.
+	NoChange ProcessErrorResult = iota
+	// ServerMarkedUnknown indicates that the error only resulted in the server being marked as Unknown.
+	ServerMarkedUnknown
+	// ConnectionPoolCleared indicates that the error resulted in the server being marked as Unknown and its connection
+	// pool being cleared.
+	ConnectionPoolCleared
+)
+
+// ServerChanged returns true if the ProcessErrorResult indicates that the server changed from an SDAM perspective
+// during a ProcessError() call.
+func (p ProcessErrorResult) ServerChanged() bool {
+	return p != NoChange
+}
+
 // ErrorProcessor implementations can handle processing errors, which may modify their internal state.
 // If this type is implemented by a Server, then Operation.Execute will call it's ProcessError
 // method after it decodes a wire message.
 type ErrorProcessor interface {
-	ProcessError(err error, conn Connection)
+	ProcessError(err error, conn Connection) ProcessErrorResult
 }
 
 // HandshakeInformation contains information extracted from a MongoDB connection handshake. This is a helper type that

x/mongo/driver/operation.go

@@ -367,7 +367,7 @@ func (op Operation) Execute(ctx context.Context, scratch []byte) error {
 		}
 		res, err = roundTrip(ctx, conn, wm)
 		if ep, ok := srvr.(ErrorProcessor); ok {
-			ep.ProcessError(err, conn)
+			_ = ep.ProcessError(err, conn)
 		}
 
 		finishedInfo.response = res

x/mongo/driver/operation_legacy.go

@@ -297,7 +297,7 @@ func (op Operation) legacyKillCursors(ctx context.Context, dst []byte, srvr Serv
 	if err != nil {
 		err = Error{Message: err.Error(), Labels: []string{TransientTransactionError, NetworkError}}
 		if ep, ok := srvr.(ErrorProcessor); ok {
-			ep.ProcessError(err, conn)
+			_ = ep.ProcessError(err, conn)
 		}
 
 		finishedInfo.cmdErr = err
@@ -635,7 +635,7 @@ func (op Operation) appendLegacyQueryDocument(dst []byte, filter bsoncore.Docume
 func (op Operation) roundTripLegacyCursor(ctx context.Context, wm []byte, srvr Server, conn Connection, collName, identifier string) (bsoncore.Document, error) {
 	wm, err := op.roundTripLegacy(ctx, conn, wm)
 	if ep, ok := srvr.(ErrorProcessor); ok {
-		ep.ProcessError(err, conn)
+		_ = ep.ProcessError(err, conn)
 	}
 	if err != nil {
 		return nil, err

x/mongo/driver/topology/sdam_spec_test.go

@@ -481,7 +481,7 @@ func applyErrors(t *testing.T, topo *Topology, errors []applicationError) {
 		case "beforeHandshakeCompletes":
 			server.ProcessHandshakeError(currError, generation)
 		case "afterHandshakeCompletes":
-			server.ProcessError(currError, &conn)
+			_ = server.ProcessError(currError, &conn)
 		default:
 			t.Fatalf("unrecognized applicationError.When value: %v", appErr.When)
 		}

x/mongo/driver/topology/server.go

@@ -356,66 +356,70 @@ func getWriteConcernErrorForProcessing(err error) (*driver.WriteConcernError, bo
 }
 
 // ProcessError handles SDAM error handling and implements driver.ErrorProcessor.
-func (s *Server) ProcessError(err error, conn driver.Connection) {
+func (s *Server) ProcessError(err error, conn driver.Connection) driver.ProcessErrorResult {
 	// ignore nil error
 	if err == nil {
-		return
+		return driver.NoChange
 	}
 
 	s.processErrorLock.Lock()
 	defer s.processErrorLock.Unlock()
 
 	// ignore stale error
 	if conn.Stale() {
-		return
+		return driver.NoChange
 	}
 	// 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.
 	desc := conn.Description()
 	if cerr, ok := err.(driver.Error); ok && (cerr.NodeIsRecovering() || cerr.NotMaster()) {
 		// ignore stale error
 		if desc.TopologyVersion.CompareToIncoming(cerr.TopologyVersion) >= 0 {
-			return
+			return driver.NoChange
 		}
 
 		// updates description to unknown
 		s.updateDescription(description.NewServerFromError(s.address, err, cerr.TopologyVersion))
 		s.RequestImmediateCheck()
 
+		res := driver.ServerMarkedUnknown
 		// If the node is shutting down or is older than 4.2, we synchronously clear the pool
 		if cerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
+			res = driver.ConnectionPoolCleared
 			s.pool.clear()
 		}
-		return
+		return res
 	}
 	if wcerr, ok := getWriteConcernErrorForProcessing(err); ok {
 		// ignore stale error
 		if desc.TopologyVersion.CompareToIncoming(wcerr.TopologyVersion) >= 0 {
-			return
+			return driver.NoChange
 		}
 
 		// updates description to unknown
 		s.updateDescription(description.NewServerFromError(s.address, err, wcerr.TopologyVersion))
 		s.RequestImmediateCheck()
 
+		res := driver.ServerMarkedUnknown
 		// If the node is shutting down or is older than 4.2, we synchronously clear the pool
 		if wcerr.NodeIsShuttingDown() || desc.WireVersion == nil || desc.WireVersion.Max < 8 {
+			res = driver.ConnectionPoolCleared
 			s.pool.clear()
 		}
-		return
+		return res
 	}
 
 	wrappedConnErr := unwrapConnectionError(err)
 	if wrappedConnErr == nil {
-		return
+		return driver.NoChange
 	}
 
 	// Ignore transient timeout errors.
 	if netErr, ok := wrappedConnErr.(net.Error); ok && netErr.Timeout() {
-		return
+		return driver.NoChange
 	}
 	if wrappedConnErr == context.Canceled || wrappedConnErr == context.DeadlineExceeded {
-		return
+		return driver.NoChange
 	}
 
 	// For a non-timeout network error, we clear the pool, set the description to Unknown, and cancel the in-progress
@@ -424,6 +428,7 @@ func (s *Server) ProcessError(err error, conn driver.Connection) {
 	s.updateDescription(description.NewServerFromError(s.address, err, nil))
 	s.pool.clear()
 	s.cancelCheck()
+	return driver.ConnectionPoolCleared
 }
 
 // update handles performing heartbeats and updating any subscribers of the