feat(alerting): classify MySQL 1236 max_allowed_packet errors separately (PeerDB-io#4417)

## Problem

MySQL error `1236 (HY000)` (`ER_MASTER_FATAL_ERROR_READING_BINLOG`) is
currently always classified as `NOTIFY_BINLOG_INVALID`. But one common
cause is a single binlog event exceeding the replica's
`max_allowed_packet`:

```
failed in pull records when: MySQL execute error: ERROR 1236 (HY000): log event entry exceeded max_allowed_packet;
Increase max_allowed_packet on source; the first event 'mysql-bin.168301' at 1789438008, ...
```

This is **not** binlog corruption/purge — the mitigation is different
(increase `max_allowed_packet` on the source, resync the affected table,
or exclude oversized blob columns), so it deserves its own error class
and tailored user notification.

## Change

- Add `ErrorNotifyBinlogEventExceededMaxAllowedPacket`
(`NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET`, action
`NotifyUser`).
- Inside `case 1236`, sub-classify on the `max_allowed_packet` message
signature; everything else still maps to `NOTIFY_BINLOG_INVALID`.
- Add a unit test covering both the new case and the generic 1236
fallback.

It closes DBI-787.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Ilia Demianenko <ilia.demianenko@clickhouse.com>

Author: 3 people

flow/alerting/classifier.go

@@ -142,6 +142,9 @@ var (
 	ErrorNotifyBinlogInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_INVALID", action: NotifyUser,
 	}
+	ErrorNotifyBinlogEventExceededMaxAllowedPacket = ErrorClass{
+		Class: "NOTIFY_BINLOG_EVENT_EXCEEDED_MAX_ALLOWED_PACKET", action: NotifyUser,
+	}
 	ErrorNotifyBinlogRowMetadataInvalid = ErrorClass{
 		Class: "NOTIFY_BINLOG_ROW_METADATA_INVALID", action: NotifyUser,
 	}
@@ -746,8 +749,13 @@ func GetErrorClass(ctx context.Context, err error) (ErrorClass, ErrorInfo) {
 			1226, // ER_USER_LIMIT_REACHED
 			1827: // ER_PASSWORD_FORMAT
 			return ErrorNotifyConnectivity, myErrorInfo
-		case 1236, // ER_MASTER_FATAL_ERROR_READING_BINLOG
-			1373: // ER_UNKNOWN_TARGET_BINLOG
+		case 1236: // ER_MASTER_FATAL_ERROR_READING_BINLOG
+			// A single binlog event larger than the replica's max_allowed_packet aborts the binlog stream read
+			if strings.Contains(myErr.Message, "max_allowed_packet") {
+				return ErrorNotifyBinlogEventExceededMaxAllowedPacket, myErrorInfo
+			}
+			return ErrorNotifyBinlogInvalid, myErrorInfo
+		case 1373: // ER_UNKNOWN_TARGET_BINLOG
 			return ErrorNotifyBinlogInvalid, myErrorInfo
 		case 1105: // ER_UNKNOWN_ERROR
 			// RDS Aurora MySQL specific errors due to "Zero Downtime Patch" or "Zero Downtime Restart"

flow/alerting/classifier_test.go

@@ -846,6 +846,38 @@ func TestAuroraMySQLZeroDowntimeRestartErrorShouldBeRecoverable(t *testing.T) {
 	}, errInfo, "Unexpected error info")
 }
 
+func TestMySQLBinlogEventExceededMaxAllowedPacket(t *testing.T) {
+	// Error 1236 caused by a binlog event larger than max_allowed_packet should be
+	// classified separately from generic binlog invalidation.
+	mysqlErr := &mysql.MyError{
+		Code:  1236, // ER_MASTER_FATAL_ERROR_READING_BINLOG
+		State: "HY000",
+		Message: "log event entry exceeded max_allowed_packet; Increase max_allowed_packet on source; " +
+			"the first event 'mysql-bin.168301' at 1789438008, the last event read from " +
+			"'/app/work2/binlogs/mysql-bin.168301' at 2333874086, the last byte read from " +
+			"'/app/work2/binlogs/mysql-bin.168301' at 2333874105.",
+	}
+	errorClass, errInfo := GetErrorClass(t.Context(), fmt.Errorf("failed in pull records: %w", mysqlErr))
+	assert.Equal(t, ErrorNotifyBinlogEventExceededMaxAllowedPacket, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "1236",
+	}, errInfo, "Unexpected error info")
+
+	// A 1236 without the max_allowed_packet signature should still be generic binlog invalidation.
+	genericErr := &mysql.MyError{
+		Code:    1236,
+		State:   "HY000",
+		Message: "Could not find first log file name in binary log index file",
+	}
+	errorClass, errInfo = GetErrorClass(t.Context(), fmt.Errorf("mysql error: %w", genericErr))
+	assert.Equal(t, ErrorNotifyBinlogInvalid, errorClass, "Unexpected error class")
+	assert.Equal(t, ErrorInfo{
+		Source: ErrorSourceMySQL,
+		Code:   "1236",
+	}, errInfo, "Unexpected error info")
+}
+
 func TestMySQLExecuteError(t *testing.T) {
 	err := exceptions.NewMySQLExecuteError(
 		tls.RecordHeaderError{Msg: "first record does not look like a TLS handshake"})