MySQL Mirror: set max execution time (#2958)

MySQL has a setting - `max_execution_time` - after which queries are
cancelled, including ours

This PR sets this setting to unlimited at a session level when
connecting to MySQL

---------

Co-authored-by: Kevin Biju <kevin@peerdb.io>

Author: Amogh-Bharadwaj

flow/activities/flowable.go

@@ -586,6 +586,7 @@ func (a *FlowableActivity) ReplicateQRepPartitions(ctx context.Context,
 		}
 
 		if err != nil {
+			logger.Error("failed to replicate partition", slog.Any("error", err))
 			return a.Alerter.LogFlowError(ctx, config.FlowJobName, err)
 		}
 	}
@@ -1239,7 +1240,7 @@ func (a *FlowableActivity) GetFlowMetadata(
 	if err != nil {
 		return nil, a.Alerter.LogFlowError(ctx, input.FlowName, err)
 	}
-	logger.Info("loaded peer types for flow", slog.String("flowName", input.FlowName),
+	logger.Debug("loaded peer types for flow", slog.String("flowName", input.FlowName),
 		slog.String("sourceName", input.SourceName), slog.String("destinationName", input.DestinationName),
 		slog.Any("peerTypes", peerTypes))
 	return &protos.FlowContextMetadata{

flow/connectors/mysql/mysql.go

@@ -182,6 +182,32 @@ func (c *MySqlConnector) connect(ctx context.Context) (*client.Conn, error) {
 		if _, err := conn.Execute("SET sql_mode = 'ANSI,NO_BACKSLASH_ESCAPES'"); err != nil {
 			return nil, fmt.Errorf("failed to set sql_mode to ANSI: %w", err)
 		}
+
+		// Set max_execution_time/max_statement_time to 0 (unlimited)
+		switch c.Flavor() {
+		case mysql.MySQLFlavor:
+			// set max_execution_time to unlimited
+			if _, err := conn.Execute("SET SESSION max_execution_time=0;"); err != nil {
+				var mErr *mysql.MyError
+				if errors.As(err, &mErr) && mErr.Code == mysql.ER_UNKNOWN_SYSTEM_VARIABLE {
+					// max_execution_time is not supported, ignore the error
+					c.logger.Warn("max_execution_time is not supported by the MySQL server, ignoring", slog.Any("error", err))
+				} else {
+					return nil, fmt.Errorf("failed to set max_execution_time to 0: %w", err)
+				}
+			}
+		case mysql.MariaDBFlavor:
+			// set max_statement_time to unlimited
+			if _, err := conn.Execute("SET SESSION max_statement_time=0;"); err != nil {
+				var mErr *mysql.MyError
+				if errors.As(err, &mErr) && mErr.Code == mysql.ER_UNKNOWN_SYSTEM_VARIABLE {
+					// max_statement_time is not supported, ignore the error
+					c.logger.Warn("max_statement_time is not supported by the MariaDB server, ignoring", slog.Any("error", err))
+				} else {
+					return nil, fmt.Errorf("failed to set max_statement_time to 0: %w", err)
+				}
+			}
+		}
 	}
 	return conn, nil
 }

flow/connectors/postgres/cdc.go

@@ -720,7 +720,6 @@ func processMessage[Items model.Items](
 			Prefix:     msg.Prefix,
 			Content:    string(msg.Content),
 		}, nil
-
 	default:
 		if _, ok := p.hushWarnUnhandledMessageType[msg.Type()]; !ok {
 			logger.Warn(fmt.Sprintf("Unhandled message type: %T", msg))