Refuse XA workloads in the binlog stream instead of applying prepared rows

.github/workflows/mysql-xa-docker.yml

@@ -0,0 +1,34 @@
+name: MySQL XA guard (isolated) /w docker-compose
+on:
+  push:
+    branches: [ main ]
+  pull_request:
+    branches: [ main ]
+permissions:
+  contents: read
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v7
+
+      - name: Detect non-docs changes
+        uses: dorny/paths-filter@ceb8a2b8f2d89434be7ff52d3de7ec3738c5cc9d # v4.0.3
+        id: filter
+        with:
+          predicate-quantifier: 'every'
+          filters: |
+            code:
+              - '!**/*.md'
+              - '!docs/**'
+
+      - name: Test XA stream guards
+        if: steps.filter.outputs.code == 'true'
+        run: docker compose -f compose.yml --profile xa up mysql xa-test --abort-on-container-exit --exit-code-from xa-test
+        working-directory: compose
+
+      - name: Skip — documentation-only change
+        if: steps.filter.outputs.code != 'true'
+        run: echo "Documentation-only change detected; skipping tests."

compose/compose.yml

@@ -29,7 +29,21 @@ services:
   test:
     build:
       context: ../
-    command: bash -c "go test -json -race -v  -timeout=10m -parallel=4  ./... | tee /proc/1/fd/1 | tparse -all"
+    # XA tests write prepare/commit events visible to every change stream on
+    # this server. Run them in xa-test's isolated CI job instead.
+    command: bash -c "go test -json -race -v -skip XA -timeout=10m -parallel=4 ./... | tee /proc/1/fd/1 | tparse -all"
+    depends_on:
+      mysql:
+        condition: service_healthy
+    environment:
+      MYSQL_DSN: tsandbox:msandbox@tcp(mysql)/test
+
+  xa-test:
+    profiles: [xa]
+    build:
+      context: ../
+    # The XA job starts its own MySQL container and runs no other tests on it.
+    command: go test -race -count=1 -timeout=10m -parallel=1 -p=1 -run XA ./...
     depends_on:
       mysql:
         condition: service_healthy

compose/replication-tls/replication-ci.yml

@@ -47,7 +47,8 @@ services:
   test:
     build:
       context: ../../.
-    command: go test -race -timeout=10m -parallel=4  ./...
+    # XA tests run on their own MySQL server in mysql-xa-docker.yml.
+    command: go test -race -skip XA -timeout=10m -parallel=4 ./...
     depends_on:
       mysql:
         condition: service_healthy

pkg/change/binlog.go

@@ -774,21 +774,30 @@ func (c *binlogClient) readStream(ctx context.Context) {
 				return
 			}
 		case *replication.QueryEvent:
-			// Query event, check if it is a DDL statement,
-			// in which case we need to notify the caller.
-			ddlTables, _, err := extractTablesFromDDLStmts(string(event.Schema), string(event.Query))
+			info, err := parseQueryEvent(string(event.Schema), string(event.Query))
 			if err != nil {
-				// The parser does not understand all syntax — the
-				// remaining classes are mode-dependent SQL (ANSI_QUOTES
-				// quoting) and syntax newer than the grammar.
-				// This behavior is copied from canal:
-				// https://github.com/go-mysql-org/go-mysql/blob/ee9447d96b48783abb05ab76a12501e5f1161e47/canal/sync.go#L144C1-L150C1
-				// We can't print the statement because it could contain user-data.
-				// We instead rely on file + pos being useful.
+				// An unparseable statement may use a SQL mode or syntax newer
+				// than the parser. Do not log the query: it may contain data.
 				c.logger.Error("Skipping query that was unable to parse", "file", currentLogName, "pos", ev.Header.LogPos)
 				continue
 			}
-			for _, ddlTable := range ddlTables {
+			// Any XA statement fails the stream: spirit does not support
+			// XA workloads. An XA transaction's row events are binlogged
+			// at XA PREPARE time, before its outcome is known — applying
+			// them treats the prepare as a commit, and a later XA ROLLBACK
+			// has no binlog representation that could undo them. "XA START"
+			// opens the group ahead of its row events, so failing here
+			// guarantees none of them are ever buffered, let alone flushed.
+			// See the matching guard in the GTID client's processQueryEvent
+			// for the full rationale and group shape.
+			if info.xa {
+				c.logger.Error("fatal error processing binlog query event", "error", errXAUnsupported)
+				c.fatalError(FatalReasonUnsupportedXA)
+				return
+			}
+			// Query event, check if it is a DDL statement,
+			// in which case we need to notify the caller.
+			for _, ddlTable := range info.tables {
 				c.processDDLNotification(ddlTable.schema, ddlTable.table)
 			}
 		case *replication.TransactionPayloadEvent:
@@ -817,7 +826,7 @@ func (c *binlogClient) readStream(ctx context.Context) {
 			}
 			if err = c.processTransactionPayload(event, eventPos); err != nil {
 				c.logger.Error("fatal error processing binlog transaction payload event", "error", err)
-				c.fatalError(FatalReasonStreamError)
+				c.fatalError(fatalReasonForStreamError(err))
 				return
 			}
 		case *replication.GTIDEvent,
@@ -831,6 +840,22 @@ func (c *binlogClient) readStream(ctx context.Context) {
 			// default can keep logging genuinely unknown event types — a future
 			// row-event variant we don't recognize could otherwise cause silent
 			// data loss.
+		case *replication.GenericEvent:
+			// Event types without a dedicated go-mysql decoder surface as
+			// GenericEvent; the header carries the real type. An
+			// XA_PREPARE_LOG_EVENT terminates an XA transaction's first
+			// binlog group (it is also how the server logs
+			// `XA COMMIT ... ONE PHASE`), and spirit does not support XA
+			// workloads. The QueryEvent guard above already fails the
+			// stream at the group's opening "XA START", before any of its
+			// row events are buffered, so this branch is defense in depth
+			// in case a future server version reshapes the group.
+			if ev.Header.EventType == replication.XA_PREPARE_LOG_EVENT {
+				c.logger.Error("fatal error processing binlog stream", "error", errXAUnsupported)
+				c.fatalError(FatalReasonUnsupportedXA)
+				return
+			}
+			c.logger.Debug("Received unknown event type", "type", ev.Header.EventType.String())
 		default:
 			c.logger.Debug("Received unknown event type", "type", fmt.Sprintf("%T", ev.Event))
 		}
@@ -1037,23 +1062,39 @@ func (c *binlogClient) processTransactionPayload(e *replication.TransactionPaylo
 				return err
 			}
 		case *replication.QueryEvent:
-			// Usually the transaction's BEGIN, which parses cleanly and
-			// yields no DDL tables. Unparseable statements are skipped the
-			// same way readStream skips them.
-			ddlTables, _, err := extractTablesFromDDLStmts(string(innerEvent.Schema), string(innerEvent.Query))
+			info, err := parseQueryEvent(string(innerEvent.Schema), string(innerEvent.Query))
 			if err != nil {
 				c.logger.Error("Skipping query inside transaction payload that was unable to parse",
 					"file", payloadPos.Name, "pos", payloadPos.Pos)
 				continue
 			}
-			for _, ddlTable := range ddlTables {
+			// XA statements fail the payload before any of its row events
+			// are buffered — see the guard in readStream's QueryEvent case.
+			// A compressed XA prepare group opens with an inner "XA START"
+			// QueryEvent, so this fires ahead of the group's RowsEvents.
+			if info.xa {
+				return errXAUnsupported
+			}
+			// Usually the transaction's BEGIN, which parses cleanly and
+			// yields no DDL tables. Unparseable statements are skipped the
+			// same way readStream skips them.
+			for _, ddlTable := range info.tables {
 				c.processDDLNotification(ddlTable.schema, ddlTable.table)
 			}
 		case *replication.TableMapEvent, *replication.XIDEvent:
 			// Housekeeping inside the payload. The TableMapEvents were
 			// already consumed by go-mysql's inner parser to decode the
 			// RowsEvents above; position tracking advances via the outer
 			// event only.
+		case *replication.GenericEvent:
+			// An inner XA_PREPARE_LOG_EVENT terminates a compressed XA
+			// prepare group. The inner "XA START" QueryEvent above already
+			// fails the payload before its row events are buffered; this is
+			// defense in depth, mirroring readStream's GenericEvent case.
+			if inner.Header.EventType == replication.XA_PREPARE_LOG_EVENT {
+				return errXAUnsupported
+			}
+			c.logger.Debug("Received unknown event type inside transaction payload", "type", inner.Header.EventType.String())
 		default:
 			// Same rationale as readStream's default case: log genuinely
 			// unknown inner event types so a future row-event variant can't