classify mysql client transient error

flow/connectors/mysql/cdc.go

@@ -467,7 +467,7 @@ func (c *MySqlConnector) PullRecords(
 			} else {
 				c.logger.Error("[mysql] PullRecords failed to get event", slog.Any("error", err))
 			}
-			return err
+			return exceptions.NewMySQLStreamingError(err)
 		}
 
 		allFetchedBytes.Add(int64(len(event.RawData)))

flow/connectors/mysql/mysql.go

@@ -24,6 +24,7 @@ import (
 	"github.com/PeerDB-io/peerdb/flow/pkg/common"
 	mysql_validation "github.com/PeerDB-io/peerdb/flow/pkg/mysql"
 	"github.com/PeerDB-io/peerdb/flow/shared"
+	"github.com/PeerDB-io/peerdb/flow/shared/exceptions"
 )
 
 type MySqlConnector struct {
@@ -299,63 +300,66 @@ func (c *MySqlConnector) ExecuteNoRetry(ctx context.Context, cmd string, args ..
 }
 
 func (c *MySqlConnector) Execute(ctx context.Context, cmd string, args ...any) (*mysql.Result, error) {
-	var connectionErr error
+	var retryableErr error
 	for conn, err := range c.withRetries(ctx) {
 		if err != nil {
 			return nil, err
 		}
 
-		rs, err := conn.Execute(cmd, args...)
-		if err != nil && mysql.ErrorEqual(err, mysql.ErrBadConn) {
-			connectionErr = err
-			continue
+		if rs, err := conn.Execute(cmd, args...); err != nil {
+			if mysql.ErrorEqual(err, mysql.ErrBadConn) || exceptions.InvalidSequenceRe.MatchString(err.Error()) {
+				retryableErr = err
+				continue
+			}
+			return nil, err
+		} else {
+			return rs, nil
 		}
-		return rs, err
 	}
-	return nil, connectionErr
+	return nil, retryableErr
 }
 
 func (c *MySqlConnector) ExecuteSelectStreaming(ctx context.Context, cmd string, result *mysql.Result,
 	rowCb client.SelectPerRowCallback,
 	resultCb client.SelectPerResultCallback,
 	args ...any,
 ) error {
-	var connectionErr error
+	var retryableErr error
 	for conn, err := range c.withRetries(ctx) {
 		if err != nil {
 			return err
 		}
 
 		if len(args) == 0 {
 			if err := conn.ExecuteSelectStreaming(cmd, result, rowCb, resultCb); err != nil {
-				if mysql.ErrorEqual(err, mysql.ErrBadConn) {
-					connectionErr = err
+				if mysql.ErrorEqual(err, mysql.ErrBadConn) || exceptions.InvalidSequenceRe.MatchString(err.Error()) {
+					retryableErr = err
 					continue
 				}
 				return err
 			}
 		} else {
 			stmt, err := conn.Prepare(cmd)
 			if err != nil {
-				if mysql.ErrorEqual(err, mysql.ErrBadConn) {
-					connectionErr = err
+				if mysql.ErrorEqual(err, mysql.ErrBadConn) || exceptions.InvalidSequenceRe.MatchString(err.Error()) {
+					retryableErr = err
 					continue
 				}
 				return err
 			}
 			err = stmt.ExecuteSelectStreaming(result, rowCb, resultCb, args...)
 			_ = stmt.Close()
 			if err != nil {
-				if mysql.ErrorEqual(err, mysql.ErrBadConn) {
-					connectionErr = err
+				if mysql.ErrorEqual(err, mysql.ErrBadConn) || exceptions.InvalidSequenceRe.MatchString(err.Error()) {
+					retryableErr = err
 					continue
 				}
 				return err
 			}
 		}
 		return nil
 	}
-	return connectionErr
+	return retryableErr
 }
 
 func (c *MySqlConnector) GetGtidModeOn(ctx context.Context) (bool, error) {

flow/shared/exceptions/mysql.go

@@ -5,11 +5,16 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"regexp"
 	"strings"
 
 	"github.com/go-mysql-org/go-mysql/mysql"
 )
 
+// InvalidSequenceRe go-mysql-org returns "invalid sequence X != Y" when the TCP packet sequence
+// is out of sync — always transient, safe to retry.
+var InvalidSequenceRe = regexp.MustCompile(`invalid sequence \d+ != \d+`)
+
 type MySQLIncompatibleColumnTypeError struct {
 	TableName  string
 	ColumnName string
@@ -72,8 +77,7 @@ func NewMySQLStreamingError(err error) *MySQLStreamingError {
 		return &MySQLStreamingError{err, true}
 	}
 
-	var recordHeaderError tls.RecordHeaderError
-	if errors.As(err, &recordHeaderError) {
+	if recordHeaderError, ok := errors.AsType[tls.RecordHeaderError](err); ok {
 		if recordHeaderError.Msg == "first record does not look like a TLS handshake" {
 			return &MySQLStreamingError{err, true}
 		}
@@ -83,6 +87,10 @@ func NewMySQLStreamingError(err error) *MySQLStreamingError {
 		return &MySQLStreamingError{err, true}
 	}
 
+	if InvalidSequenceRe.MatchString(err.Error()) {
+		return &MySQLStreamingError{err, true}
+	}
+
 	return &MySQLStreamingError{err, false}
 }