Classify wal segment removed as notify user (#4107)

From prod logs, this error either shows up once then recovers, or
persists until user resyncs. Nothing we can do to help, so classify as
notify_user

Author: ilidemi

flow/alerting/classifier.go

@@ -221,6 +221,9 @@ var (
 	ErrorNotifyPostgresLogicalMessageProcessing = ErrorClass{
 		Class: "NOTIFY_POSTGRES_LOGICAL_MESSAGE_PROCESSING_ERROR", action: NotifyUser,
 	}
+	ErrorNotifyWalSegmentRemoved = ErrorClass{
+		Class: "NOTIFY_WAL_SEGMENT_REMOVED", action: NotifyUser,
+	}
 	ErrorNotifyClickHouseSupportIsDisabledError = ErrorClass{
 		Class: "NOTIFY_CLICKHOUSE_SUPPORT_IS_DISABLED_ERROR", action: NotifyUser,
 	}
@@ -527,9 +530,12 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 
 		case pgerrcode.UndefinedFile:
 			// Handle WAL segment removed errors
-			// There is a quirk in some PG installs where replication can try read a segment that hasn't been created yet but will show up
+			// It either shows up once then disappears
+			// (quirk in some PG installs where replication can try read a segment that hasn't been created yet)
+			// or shows up and persists
+			// NotifyUser with repeat threshold accommodates both
 			if PostgresWalSegmentRemovedRe.MatchString(pgErr.Message) {
-				return ErrorRetryRecoverable, pgErrorInfo
+				return ErrorNotifyWalSegmentRemoved, pgErrorInfo
 			}
 			// Handles missing spill-to-disk file during logical decoding (transient error)
 			if PostgresSpillFileMissingRe.MatchString(pgErr.Message) {
@@ -557,9 +563,12 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			}
 
 			// Handle WAL segment removed errors
-			// There is a quirk in some PG installs where replication can try read a segment that hasn't been created yet but will show up
+			// It either shows up once then disappears
+			// (quirk in some PG installs where replication can try read a segment that hasn't been created yet)
+			// or shows up and persists
+			// NotifyUser with repeat threshold accommodates both
 			if PostgresWalSegmentRemovedRe.MatchString(pgErr.Message) {
-				return ErrorRetryRecoverable, pgErrorInfo
+				return ErrorNotifyWalSegmentRemoved, pgErrorInfo
 			}
 
 			// Handle Neon quota exceeded errors

flow/alerting/classifier_test.go

@@ -107,21 +107,25 @@ func TestClickHouseSelectFromDestinationDuringQrepAsMvError(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
-func TestPostgresWalRemovedErrorShouldBeRecoverable(t *testing.T) {
-	// Simulate a WAL removed error
-	err := &exceptions.PostgresWalError{
-		Msg: &pgproto3.ErrorResponse{
-			Severity: "ERROR",
-			Code:     pgerrcode.InternalError,
-			Message:  "requested WAL segment 000000010001337F0000002E has already been removed",
-		},
+func TestPostgresWalRemovedErrorShouldBeNotifyUser(t *testing.T) {
+	for _, code := range []string{pgerrcode.InternalError, pgerrcode.UndefinedFile} {
+		t.Run(code, func(t *testing.T) {
+			// Simulate a WAL removed error
+			err := &exceptions.PostgresWalError{
+				Msg: &pgproto3.ErrorResponse{
+					Severity: "ERROR",
+					Code:     code,
+					Message:  "requested WAL segment 000000010001337F0000002E has already been removed",
+				},
+			}
+			errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("error in WAL: %w", err))
+			assert.Equal(t, ErrorNotifyWalSegmentRemoved, errorClass, "Unexpected error class")
+			assert.Equal(t, ErrorInfo{
+				Source: ErrorSourcePostgres,
+				Code:   code,
+			}, errInfo, "Unexpected error info")
+		})
 	}
-	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("error in WAL: %w", err))
-	assert.Equal(t, ErrorRetryRecoverable, errorClass, "Unexpected error class")
-	assert.Equal(t, ErrorInfo{
-		Source: ErrorSourcePostgres,
-		Code:   pgerrcode.InternalError,
-	}, errInfo, "Unexpected error info")
 }
 
 func TestAuroraInternalWALErrorShouldBeRecoverable(t *testing.T) {