mongo error clean up (#3591)

`changestream.Err()` calls
[`wrapErrors`](https://github.com/mongodb/mongo-go-driver/blob/1e7cea0fc96b08059915a3d1b0903be2c7c1ed0c/mongo/change_stream.go#L569-L579)
which [_attempts_ to wraps errors in one of CommandError, MarshalError,
MongocryptError,
etc.](https://github.com/mongodb/mongo-go-driver/blob/1e7cea0fc96b08059915a3d1b0903be2c7c1ed0c/mongo/errors.go#L90-L106),
but errors can fall thru if it doesn't match one of the known error
targets, and the original error is returned. This PR updates
driver.Error matching with the wrapped errors.

There's also been a few iterations of this error handling such that this
section no longer makes sense (comment says we retry, but code has since
been changed to notify, but notify to all retryable errors seems too
aggressive):
```
		// some error codes are defined to be retryable by the driver, so we retry them
		var retryableError RetryableError
		if errors.As(mongoErr, &retryableError) && retryableError.Retryable() {
			return ErrorNotifyConnectivity, mongoErrorInfo
		}
```

So instead notify on the errors we've seen so far: ShutdownInProgress
(which has code=91). Would be better if we are more explicit about which
one to retry and which one to notify, as they come up in practice.

Finally, fixes
https://linear.app/clickhouse/issue/DBI-111/classify-connection-pool-error-with-mongodb-cluster
by making it retryable when "connection reset by peers" is detected.

Author: jgao54

flow/alerting/classifier.go

@@ -16,6 +16,7 @@ import (
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/jackc/pgerrcode"
 	"github.com/jackc/pgx/v5/pgconn"
+	"go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver/topology"
 	"go.temporal.io/sdk/temporal"
@@ -92,11 +93,6 @@ type ErrorClass struct {
 	action ErrorAction
 }
 
-type RetryableError interface {
-	error
-	Retryable() bool
-}
-
 var (
 	ErrorNotifyDestinationModified = ErrorClass{
 		Class: "NOTIFY_DESTINATION_MODIFIED", action: NotifyUser,
@@ -610,41 +606,57 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
-	var mongoErr driver.Error
-	if errors.As(err, &mongoErr) {
-		// https://www.mongodb.com/docs/manual/reference/error-codes/
+	var mongoCmdErr mongo.CommandError
+	if errors.As(err, &mongoCmdErr) {
 		mongoErrorInfo := ErrorInfo{
 			Source: ErrorSourceMongoDB,
-			Code:   strconv.Itoa(int(mongoErr.Code)),
+			Code:   strconv.Itoa(int(mongoCmdErr.Code)),
 		}
 
-		if mongoErr.RetryableRead() {
-			return ErrorNotifyConnectivity, mongoErrorInfo
+		if mongoCmdErr.HasErrorMessage("connection reset by peer") {
+			return ErrorRetryRecoverable, mongoErrorInfo
 		}
 
-		// some driver errors do not provide error code, such as `poolClearedError`, so we check label instead
-		for _, label := range mongoErr.Labels {
-			if label == driver.TransientTransactionError {
-				return ErrorNotifyConnectivity, mongoErrorInfo
-			}
+		// this often happens on Mongo Atlas as part of maintenance, and should recover, but we notify if exceed default threshold
+		// (ShutdownInProgress code should be 91, but we have observed 0 in the past, so string match to be safe)
+		if mongoCmdErr.HasErrorMessage("(ShutdownInProgress) The server is in quiesce mode and will shut down") {
+			return ErrorNotifyConnectivity, mongoErrorInfo
 		}
 
-		// some error codes are defined to be retryable by the driver, so we retry them
-		var retryableError RetryableError
-		if errors.As(mongoErr, &retryableError) && retryableError.Retryable() {
+		// This should recover, but we notify if exceed default threshold
+		if mongoCmdErr.HasErrorLabel(driver.TransientTransactionError) {
 			return ErrorNotifyConnectivity, mongoErrorInfo
 		}
 
-		switch mongoErr.Code {
+		// https://www.mongodb.com/docs/manual/reference/error-codes/
+		switch mongoCmdErr.Code {
 		case 13: // Unauthorized
 			return ErrorNotifyConnectivity, mongoErrorInfo
+		case 91: // ShutdownInProgress
+			return ErrorNotifyConnectivity, mongoErrorInfo
 		case 286: // ChangeStreamHistoryLost
 			return ErrorNotifyChangeStreamHistoryLost, mongoErrorInfo
 		default:
 			return ErrorOther, mongoErrorInfo
 		}
 	}
 
+	var mongoMarshalErr mongo.MarshalError
+	if errors.As(err, &mongoMarshalErr) {
+		return ErrorOther, ErrorInfo{
+			Source: ErrorSourceMongoDB,
+			Code:   "MARSHAL_ERROR",
+		}
+	}
+
+	var mongoEncryptError mongo.MongocryptError
+	if errors.As(err, &mongoEncryptError) {
+		return ErrorOther, ErrorInfo{
+			Source: ErrorSourceMongoDB,
+			Code:   "MONGOCRYPT_ERROR",
+		}
+	}
+
 	var mongoServerError topology.ServerSelectionError
 	if errors.As(err, &mongoServerError) {
 		return ErrorNotifyConnectivity, ErrorInfo{
@@ -653,6 +665,14 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	var mongoConnError topology.ConnectionError
+	if errors.As(err, &mongoConnError) {
+		return ErrorNotifyConnectivity, ErrorInfo{
+			Source: ErrorSourceMongoDB,
+			Code:   "CONNECTION_ERROR",
+		}
+	}
+
 	var chException *clickhouse.Exception
 	if errors.As(err, &chException) {
 		chErrorInfo := ErrorInfo{

flow/alerting/classifier_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/jackc/pgx/v5/pgproto3"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	"go.mongodb.org/mongo-driver/v2/mongo"
 	"go.mongodb.org/mongo-driver/v2/x/mongo/driver"
 	"go.temporal.io/sdk/temporal"
 
@@ -574,30 +575,20 @@ func TestTemporalUnknownErrorShouldBeOther(t *testing.T) {
 
 func TestMongoShutdownInProgressErrorShouldBeRecoverable(t *testing.T) {
 	// Simulate a MongoDB shutdown in progress error (quiesce mode)
-	err := driver.Error{
-		Message: "connection pool for <host>:<port> was cleared because another operation failed with",
-		Labels:  []string{driver.TransientTransactionError},
-		Wrapped: errors.New("the server is in quiesce mode and will shut down"),
+	de := driver.Error{
+		Code: 0,
+		//nolint:lll
+		Message: "connection pool for <host>:<port> was cleared because another operation failed with: (ShutdownInProgress) The server is in quiesce mode and will shut down",
 	}
-	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("change stream error: %w", err))
-	assert.Equal(t, ErrorNotifyConnectivity, errorClass, "Unexpected error class")
-	assert.Equal(t, ErrorInfo{
-		Source: ErrorSourceMongoDB,
-		Code:   "0",
-	}, errInfo, "Unexpected error info")
-}
-
-func TestMongoUnauthorizedErrorShouldBeConnectivity(t *testing.T) {
-	// Simulate a MongoDB unauthorized error
-	err := driver.Error{
-		Code:    13,
-		Message: "Command getMore requires authentication",
-		Name:    "Unauthorized",
+	err := mongo.CommandError{
+		Message: de.Message,
+		Code:    de.Code,
+		Wrapped: de,
 	}
 	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("change stream error: %w", err))
 	assert.Equal(t, ErrorNotifyConnectivity, errorClass, "Unexpected error class")
 	assert.Equal(t, ErrorInfo{
 		Source: ErrorSourceMongoDB,
-		Code:   "13",
+		Code:   "0",
 	}, errInfo, "Unexpected error info")
 }