classify changestream creation error

Author: jgao54

flow/alerting/classifier.go

@@ -63,6 +63,7 @@ var (
 	PostgresSpillFileMissingRe        = regexp.MustCompile(`Unable to restore changes for xid \d+`)
 	MySqlRdsBinlogFileNotFoundRe      = regexp.MustCompile(`File '/rdsdbdata/log/binlog/mysql-bin-changelog.\d+' not found`)
 	MongoPoolClearedErrorRe           = regexp.MustCompile(`connection pool for .+ was cleared because another operation failed with`)
+	MongoServerSideTimeoutRe          = regexp.MustCompile(`calculated server-side timeout \(.*\) is less than or equal to \d+`)
 )
 
 func (e ErrorAction) String() string {
@@ -221,6 +222,10 @@ var (
 	ErrorNotifyChangeStreamHistoryLost = ErrorClass{
 		Class: "NOTIFY_CHANGE_STREAM_HISTORY_LOST", action: NotifyUser,
 	}
+	// ErrorNotifyMongoServerSideTimeout fires when the Mongo driver rejects Watch because the computed maxTimeMS is <= 0
+	ErrorNotifyMongoServerSideTimeout = ErrorClass{
+		Class: "NOTIFY_MONGO_INVALID_TIMEOUT", action: NotifyUser,
+	}
 	ErrorNotifyPostgresLogicalMessageProcessing = ErrorClass{
 		Class: "NOTIFY_POSTGRES_LOGICAL_MESSAGE_PROCESSING_ERROR", action: NotifyUser,
 	}
@@ -1136,6 +1141,19 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 		}
 	}
 
+	if mongoCreateChangeStreamErr, ok := errors.AsType[*exceptions.MongoCreateChangeStreamError](err); ok {
+		if MongoServerSideTimeoutRe.MatchString(mongoCreateChangeStreamErr.Error()) {
+			return ErrorNotifyMongoServerSideTimeout, ErrorInfo{
+				Source: ErrorSourceMongoDB,
+				Code:   "SERVER_SIDE_TIMEOUT",
+			}
+		}
+		return ErrorOther, ErrorInfo{
+			Source: ErrorSourceMongoDB,
+			Code:   "UNKNOWN",
+		}
+	}
+
 	return ErrorOther, ErrorInfo{
 		Source: ErrorSourceOther,
 		Code:   "UNKNOWN",

flow/alerting/classifier_test.go

@@ -757,6 +757,20 @@ func TestMongoPoolErrorShouldBeRecoverable(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestMongoCreateChangeStreamServerSideTimeoutShouldNotifyUser(t *testing.T) {
+	e := errors.New("calculated server-side timeout (0 ms) is less than or equal to 0" +
+		"(network round-trip time stats: moving avg: 20.14653ms, min: 20.054326ms, moving " +
+		"stddev: 105.47µs): operation not sent to server, as Timeout would be exceeded: " +
+		"context deadline exceeded")
+	err := exceptions.NewMongoCreateChangeStreamError(e)
+	errorClass, errInfo := GetErrorClass(t.Context(), err)
+	assert.Equal(t, ErrorNotifyMongoServerSideTimeout, errorClass)
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMongoDB,
+		Code:   "SERVER_SIDE_TIMEOUT",
+	}, errInfo)
+}
+
 func TestMongoCursorErrors(t *testing.T) {
 	err := mongo.CommandError{
 		Code:    6,

flow/connectors/mongo/cdc.go

@@ -476,7 +476,11 @@ func createChangeStream(
 ) (*mongo.ChangeStream, error) {
 	watchCtx, cancel := context.WithTimeout(parent, 5*time.Minute)
 	defer cancel()
-	return client.Watch(watchCtx, pipeline, opts)
+	cs, err := client.Watch(watchCtx, pipeline, opts)
+	if err != nil {
+		return nil, exceptions.NewMongoCreateChangeStreamError(err)
+	}
+	return cs, nil
 }
 
 // This can happen if the resumeToken we are attempting to `ResumeAfter` refers to a table that has been

flow/shared/exceptions/mongo.go

@@ -15,3 +15,19 @@ func NewInvalidIdValueError(table string) *MongoInvalidIdValueError {
 func (e *MongoInvalidIdValueError) Error() string {
 	return fmt.Sprintf("_id field is missing or null in table %s; _id must be present and non-null", e.Table)
 }
+
+type MongoCreateChangeStreamError struct {
+	error
+}
+
+func NewMongoCreateChangeStreamError(err error) *MongoCreateChangeStreamError {
+	return &MongoCreateChangeStreamError{err}
+}
+
+func (e *MongoCreateChangeStreamError) Error() string {
+	return e.error.Error()
+}
+
+func (e *MongoCreateChangeStreamError) Unwrap() error {
+	return e.error
+}