From 3fff729d50b7f997012957da3b923422ec8154e5 Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 23 Jul 2026 19:03:53 -0600 Subject: [PATCH 1/4] Refuse XA workloads in the binlog stream instead of applying prepared rows Both change clients treated an XA transaction's prepare-time row events as committed: the row images are written to the binary log at XA PREPARE time and were buffered and flushed to the target like any commit. If the transaction was later terminated with XA ROLLBACK, nothing in the binlog undoes those rows, so the target diverged permanently - detectable only by checksum, and only if the affected chunk was checksummed after the rollback. Full XA support (tracking prepared XIDs and buffering until the XA COMMIT / XA ROLLBACK outcome) is out of scope, so fail fast instead, matching the existing posture of refusing binlog_row_value_options: any XA statement (XA START / XA END / XA COMMIT / XA ROLLBACK QueryEvents) or XA_PREPARE_LOG_EVENT observed in the stream fails the migration with "XA transactions detected in the binlog stream: spirit does not support XA workloads", surfaced through the existing fatal stream-error path (checkpoint preserved). The guard fires at the group's opening "XA START" QueryEvent, which the server writes ahead of the transaction's row events, so none of them are ever buffered - a concurrent flush therefore cannot apply them, and the resume coordinate never advances past the refused group. Applies to both the GTID and file/offset clients, for uncompressed streams and binlog_transaction_compression payloads alike. Co-Authored-By: Claude Fable 5 --- pkg/change/binlog.go | 46 ++++ pkg/change/binlog_test.go | 161 +++++++++++++ pkg/change/gtid.go | 141 ++++++------ pkg/change/gtid_test.go | 464 ++++++++++++++++++++++++-------------- pkg/change/utils.go | 27 +++ 5 files changed, 600 insertions(+), 239 deletions(-) diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go index 5e1e1312f..016aa8a5f 100644 --- a/pkg/change/binlog.go +++ b/pkg/change/binlog.go @@ -628,6 +628,20 @@ func (c *binlogClient) readStream(ctx context.Context) { return } case *replication.QueryEvent: + // 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 isXAStatement(strings.TrimSpace(string(event.Query))) { + c.logger.Error("fatal error processing binlog query event", "error", errXAUnsupported) + c.fatalError(FatalReasonStreamError) + return + } // 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)) @@ -685,6 +699,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(FatalReasonStreamError) + 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)) } @@ -870,6 +900,13 @@ func (c *binlogClient) processTransactionPayload(e *replication.TransactionPaylo return err } case *replication.QueryEvent: + // 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 isXAStatement(strings.TrimSpace(string(innerEvent.Query))) { + 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. @@ -887,6 +924,15 @@ func (c *binlogClient) processTransactionPayload(e *replication.TransactionPaylo // 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 diff --git a/pkg/change/binlog_test.go b/pkg/change/binlog_test.go index 85f8bd083..ecb1ab65f 100644 --- a/pkg/change/binlog_test.go +++ b/pkg/change/binlog_test.go @@ -6,6 +6,7 @@ import ( "errors" "fmt" "log/slog" + "strings" "sync" "sync/atomic" "testing" @@ -18,7 +19,9 @@ import ( "github.com/block/spirit/pkg/testutils" "github.com/block/spirit/pkg/utils" "github.com/go-mysql-org/go-mysql/mysql" + "github.com/go-mysql-org/go-mysql/replication" mysql2 "github.com/go-sql-driver/mysql" + "github.com/google/uuid" "github.com/stretchr/testify/require" "go.uber.org/goleak" ) @@ -709,6 +712,164 @@ func TestDDLNotificationTransactionCompression(t *testing.T) { require.Equal(t, FatalReasonSchemaChange, <-cancelled) } +// TestBinlogClientXATransactionGuard is the binlog (file/offset) client's +// twin of TestGTIDClientXATransaction: the non-GTID path buffers and +// applies XA prepare-time row images exactly the same way, so it carries +// the same guard. A real two-phase XA transaction's first group — +// Query("XA START ..."), the row events, Query("XA END ...") and the +// terminating XA_PREPARE_LOG_EVENT — is written to the binlog in one +// burst at XA PREPARE time; the guard must fail the stream at the +// opening "XA START" QueryEvent, before any of the row events after it +// are buffered, and classify the abort as a checkpoint-preserving +// stream error. Unique xids and per-run table names for the reasons +// documented on TestGTIDClientXATransaction. +func TestBinlogClientXATransactionGuard(t *testing.T) { + db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) + require.NoError(t, err) + defer utils.CloseAndLog(db) + + xid := xaTestXIDPrefix + "_binlog_" + uuid.NewString() + suffix := strings.ReplaceAll(uuid.NewString(), "-", "")[:8] + srcTable := "binlogxat1_" + suffix + dstTable := "binlogxat2_" + suffix + + // Best-effort sweep of stragglers from prior interrupted runs (a + // logged no-op without XA_RECOVER_ADMIN). + rollbackDanglingXATestTxns(t, db) + testutils.RunSQL(t, fmt.Sprintf("DROP TABLE IF EXISTS %s, %s", srcTable, dstTable)) + testutils.RunSQL(t, fmt.Sprintf("CREATE TABLE %s (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))", srcTable)) + testutils.RunSQL(t, fmt.Sprintf("CREATE TABLE %s (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))", dstTable)) + t.Cleanup(func() { + // See TestGTIDClientXATransaction: terminate this run's XA + // transactions before dropping the tables, from a fresh connection. + cleanupDB, err := sql.Open("mysql", testutils.DSN()) + if err != nil { + return + } + defer utils.CloseAndLog(cleanupDB) + rollbackDanglingXATestTxns(t, cleanupDB, xid) + _, _ = cleanupDB.ExecContext(context.Background(), "SET SESSION lock_wait_timeout=5") + _, _ = cleanupDB.ExecContext(context.Background(), fmt.Sprintf("DROP TABLE IF EXISTS %s, %s", srcTable, dstTable)) + }) + + t1 := table.NewTableInfo(db, "test", srcTable) + require.NoError(t, t1.SetInfo(t.Context())) + t2 := table.NewTableInfo(db, "test", dstTable) + require.NoError(t, t2.SetInfo(t.Context())) + + cfg, err := mysql2.ParseDSN(testutils.DSN()) + require.NoError(t, err) + + var gotReason atomic.Int64 + gotReason.Store(-1) + clientConfig := NewClientDefaultConfig() + clientConfig.CancelFunc = func(reason FatalReason) bool { + gotReason.Store(int64(reason)) + return true + } + client := NewBinlogClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), clientConfig).(*binlogClient) + chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) + require.NoError(t, err) + require.NoError(t, client.AddSubscription(t1, t2, chunker)) + require.NoError(t, client.Start(t.Context())) + defer client.Close() + + // XA transactions are session-scoped between XA START and XA + // PREPARE: pin a single connection for the whole two-phase dance. + conn, err := db.Conn(t.Context()) + require.NoError(t, err) + defer utils.CloseAndLog(conn) + xaExec := func(stmt string) { + t.Helper() + _, err := conn.ExecContext(t.Context(), stmt) + require.NoError(t, err) + } + + xaExec(fmt.Sprintf("XA START '%s'", xid)) + xaExec(fmt.Sprintf("INSERT INTO %s (a, b, c) VALUES (1, 2, 3)", srcTable)) + xaExec(fmt.Sprintf("INSERT INTO %s (a, b, c) VALUES (2, 3, 4)", srcTable)) + xaExec(fmt.Sprintf("XA END '%s'", xid)) + + // Before XA PREPARE nothing of the transaction exists in the binary + // log, so no row events can have streamed and the guard cannot have + // fired. + require.NoError(t, client.BlockWait(t.Context())) + require.Equal(t, 0, client.GetDeltaLen(), "row events must not stream before XA PREPARE") + require.Equal(t, int64(-1), gotReason.Load(), "the guard must not fire before the XA group is binlogged") + + xaExec(fmt.Sprintf("XA PREPARE '%s'", xid)) + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, + 5*time.Second, 5*time.Millisecond, "XA PREPARE must fail the stream as a stream error") + client.streamWG.Wait() // reader fully exited: buffering is final + require.Equal(t, 0, client.GetDeltaLen(), "no prepared row events may be buffered once the guard fires") + + // Roll the prepared transaction back and verify the target never saw + // the prepared rows — the divergence the guard exists to prevent. + xaExec(fmt.Sprintf("XA ROLLBACK '%s'", xid)) + var count int + require.NoError(t, db.QueryRowContext(t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstTable)).Scan(&count)) + require.Equal(t, 0, count, "prepared-then-rolled-back rows must never reach the target") +} + +// TestBinlogProcessTransactionPayloadXAGuard is the binlog client's twin +// of TestGTIDProcessTransactionPayloadXAGuard: an inner "XA START" +// QueryEvent must fail a compressed payload before the row events after +// it are buffered, and an inner XA_PREPARE_LOG_EVENT (the +// defense-in-depth branch) must fail it too. +func TestBinlogProcessTransactionPayloadXAGuard(t *testing.T) { + db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) + require.NoError(t, err) + defer utils.CloseAndLog(db) + + // Real tables so the subscription would buffer the inner row event + // if the guard failed to fire first. + testutils.RunSQL(t, "DROP TABLE IF EXISTS binlogxapayt1, binlogxapayt2") + testutils.RunSQL(t, "CREATE TABLE binlogxapayt1 (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))") + testutils.RunSQL(t, "CREATE TABLE binlogxapayt2 (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))") + + t1 := table.NewTableInfo(db, "test", "binlogxapayt1") + require.NoError(t, t1.SetInfo(t.Context())) + t2 := table.NewTableInfo(db, "test", "binlogxapayt2") + require.NoError(t, t2.SetInfo(t.Context())) + + cfg, err := mysql2.ParseDSN(testutils.DSN()) + require.NoError(t, err) + client := NewBinlogClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), NewClientDefaultConfig()).(*binlogClient) + chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) + require.NoError(t, err) + require.NoError(t, client.AddSubscription(t1, t2, chunker)) + defer client.Close() + + // A compressed XA prepare group as go-mysql decompresses it: the + // inner "XA START" QueryEvent precedes the inner row events. + xaGroup := &replication.TransactionPayloadEvent{Events: []*replication.BinlogEvent{ + { + Header: &replication.EventHeader{EventType: replication.QUERY_EVENT}, + Event: &replication.QueryEvent{Schema: []byte("test"), Query: []byte("XA START X'78',X'',1")}, + }, + { + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("test"), Table: []byte("binlogxapayt1")}, + Rows: [][]any{{int32(1), int32(0), int32(0)}}, + }, + }, + }} + require.ErrorIs(t, client.processTransactionPayload(xaGroup, mysql.Position{Name: "binlog.000001", Pos: 4}), errXAUnsupported) + require.Equal(t, 0, client.GetDeltaLen(), "the inner row events after XA START must not be buffered") + + // Defense in depth: an inner XA_PREPARE_LOG_EVENT without its + // opening "XA START". + prepareOnly := &replication.TransactionPayloadEvent{Events: []*replication.BinlogEvent{ + { + Header: &replication.EventHeader{EventType: replication.XA_PREPARE_LOG_EVENT}, + Event: &replication.GenericEvent{}, + }, + }} + require.ErrorIs(t, client.processTransactionPayload(prepareOnly, mysql.Position{Name: "binlog.000001", Pos: 4}), errXAUnsupported) + require.Equal(t, 0, client.GetDeltaLen()) +} + // TestCompositePKUpdate tests that we correctly handle // the case when a PRIMARY KEY is moved. // See: https://github.com/block/spirit/issues/417 diff --git a/pkg/change/gtid.go b/pkg/change/gtid.go index c716f7974..5d6742f45 100644 --- a/pkg/change/gtid.go +++ b/pkg/change/gtid.go @@ -214,10 +214,9 @@ func (c *gtidClient) getBufferedGTID() mysql.GTIDSet { // // This must be called when — and only when — the current transaction's // binlog group ends: XIDEvent (InnoDB commit), a COMMIT or ROLLBACK -// QueryEvent (non-transactional engines / mixed-engine rollbacks), a +// QueryEvent (non-transactional engines / mixed-engine rollbacks), or a // standalone statement that is its own transaction (DDL, statements the -// TiDB parser cannot parse such as CREATE TRIGGER or stored procedures), -// an XA_PREPARE_LOG_EVENT, or an XA COMMIT / XA ROLLBACK QueryEvent. +// TiDB parser cannot parse such as CREATE TRIGGER or stored procedures). // Every GTIDEvent the server streams corresponds to an entry in its // gtid_executed, so any path that drops a pending GTID instead of // promoting it leaves bufferedGTID permanently behind gtid_executed and @@ -229,19 +228,10 @@ func (c *gtidClient) getBufferedGTID() mysql.GTIDSet { // and its row events are silently lost. This is why a mid-group QueryEvent // must not promote. Regular transactions have no mid-group QueryEvents // besides BEGIN and the SAVEPOINT family ("SAVEPOINT `x`", plus -// "ROLLBACK TO `x`" in mixed-engine transactions), but XA transactions -// do — their first group is written to -// the binlog in one piece at XA PREPARE time, shaped as (verified against -// MySQL 8.0): -// -// GTIDEvent(g1) → Query("XA START x") → row events → -// Query("XA END x") → XA_PREPARE_LOG_EVENT -// -// with the terminal XA COMMIT or XA ROLLBACK arriving any amount of time -// later as a QueryEvent under its own GTID (g2), with no row events. -// (`XA COMMIT ... ONE PHASE` is instead logged as the XA_PREPARE_LOG_EVENT -// terminator of the first group, so both group shapes end at either an -// XA_PREPARE_LOG_EVENT or a plain QueryEvent terminator.) +// "ROLLBACK TO `x`" in mixed-engine transactions). XA transactions log +// mid-group QueryEvents too ("XA START"/"XA END"), but any XA statement +// now fails the stream outright instead of adjusting pending-GTID state — +// see the XA guard in processQueryEvent. func (c *gtidClient) promotePendingGTID() { c.mu.Lock() pendingSID := c.pendingSID @@ -573,7 +563,11 @@ func (c *gtidClient) readStream(ctx context.Context) { return } case *replication.QueryEvent: - c.processQueryEvent(event) + if err = c.processQueryEvent(event); err != nil { + c.logger.Error("fatal error processing GTID query event", "error", err) + c.fatalError(FatalReasonStreamError) + return + } case *replication.TransactionPayloadEvent: // binlog_transaction_compression=ON wraps the whole transaction // (BEGIN QueryEvent, TableMapEvents, row events, XIDEvent) in one @@ -600,14 +594,16 @@ func (c *gtidClient) readStream(ctx context.Context) { // GenericEvent; the header carries the real type. The one we // must act on is XA_PREPARE_LOG_EVENT: it terminates an XA // transaction's first binlog group (it is also how the server - // logs `XA COMMIT ... ONE PHASE`). All of the transaction's row - // events precede it in the group and have been buffered, and - // the server records the GTID in gtid_executed at prepare time, - // so this — not the earlier "XA START"/"XA END" QueryEvents — - // is the point where the pending GTID is safe to promote. + // logs `XA COMMIT ... ONE PHASE`), and spirit does not support + // XA workloads — see the guard in processQueryEvent. That guard + // already fails the stream at the group's opening "XA START" + // QueryEvent, 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.promotePendingGTID() - continue + c.logger.Error("fatal error processing GTID stream", "error", errXAUnsupported) + c.fatalError(FatalReasonStreamError) + return } c.logger.Debug("Received unknown event type", "type", ev.Header.EventType.String()) default: @@ -618,17 +614,20 @@ func (c *gtidClient) readStream(ctx context.Context) { // processQueryEvent handles a QueryEvent, whether read directly from the // stream or decompressed from a transaction payload. Transaction-control -// statements adjust the pending-GTID state; everything else goes through -// DDL extraction. See promotePendingGTID for the group shapes that dictate -// which statements promote and which must leave the pending GTID pending. -func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { +// statements adjust the pending-GTID state, XA statements fail the stream +// (spirit does not support XA workloads; see the guard below), and +// everything else goes through DDL extraction. See promotePendingGTID for +// the group shapes that dictate which statements promote and which must +// leave the pending GTID pending. A returned error is fatal: the caller +// must tear the stream down without buffering anything further. +func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { // A "BEGIN" QueryEvent inside a transaction is not DDL — skip // it cheaply rather than handing it to the parser. The pending // GTID must stay pending: the transaction's row events have not // been buffered yet. q := strings.TrimSpace(string(event.Query)) if strings.EqualFold(q, "BEGIN") { - return + return nil } // MySQL also logs SAVEPOINT statements as QueryEvents in the // *middle* of a row-format transaction (verified against MySQL @@ -654,7 +653,7 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { // does not binlog it today) since releasing a savepoint never // ends a transaction either. if hasPrefixFold(q, "SAVEPOINT ") || hasPrefixFold(q, "ROLLBACK TO ") || hasPrefixFold(q, "RELEASE SAVEPOINT ") { - return + return nil } // COMMIT/ROLLBACK QueryEvents end a transaction that involved a // non-transactional engine (these get a QueryEvent terminator @@ -666,33 +665,36 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { // promotion here would wedge BlockWait forever. if strings.EqualFold(q, "COMMIT") || strings.EqualFold(q, "ROLLBACK") { c.promotePendingGTID() - return - } - // XA statements must not fall through to the parser path below - // (which promotes): see the XA group shape documented on - // promotePendingGTID. "XA START" opens the group exactly like - // BEGIN — its row events have not been buffered yet — and - // "XA END" is not a terminator either (the group ends at the - // XA_PREPARE_LOG_EVENT that follows), so for both the pending - // GTID must stay pending. Promoting here would let a concurrent - // flush publish g1 as a resume coordinate before its row events - // are buffered; a crash before the next flush then resumes past - // g1 and silently loses its rows. The server rewrites these - // statements canonically (`XA BEGIN 'x'` is binlogged as - // "XA START X'78',X'',1"), so a keyword prefix match is exact. - if hasPrefixFold(q, "XA START ") || hasPrefixFold(q, "XA END ") { - return + return nil } - // XA COMMIT / XA ROLLBACK (the two-phase outcome, decided after - // the prepare) are each their own single-statement transaction: - // own GTID, no row events. Promote, same as COMMIT/ROLLBACK - // above. (The one-phase variant `XA COMMIT ... ONE PHASE` never - // takes this path — it is logged as an XA_PREPARE_LOG_EVENT, - // handled by readStream's GenericEvent case for uncompressed - // groups and by processTransactionPayload for compressed ones.) - if hasPrefixFold(q, "XA COMMIT ") || hasPrefixFold(q, "XA ROLLBACK ") { - c.promotePendingGTID() - return + // Any XA statement fails the stream: spirit does not support XA + // workloads. An XA transaction's first binlog group is written in + // one piece at XA PREPARE time (verified against MySQL 8.0): + // + // GTIDEvent(g1) → Query("XA START x") → row events → + // Query("XA END x") → XA_PREPARE_LOG_EVENT + // + // with the terminal XA COMMIT or XA ROLLBACK arriving any amount + // of time later as a QueryEvent under its own GTID (g2), with no + // row events. The row events are therefore streamed before the + // transaction's outcome is known: buffering and flushing them + // treats the prepare as a commit, and a later XA ROLLBACK has no + // binlog representation that could undo them — the target would + // diverge permanently, detectable only by checksum. Rather than + // track prepared XIDs and buffer until the outcome, refuse the + // workload. Failing on "XA START" — before any of the group's row + // events — guarantees none of them are ever buffered, let alone + // flushed. A terminal XA COMMIT / XA ROLLBACK with no preceding + // "XA START" in-stream means the transaction was prepared before + // we connected: its row events were never streamed and may postdate + // the copier's snapshot of their chunk, so an XA COMMIT outcome + // could silently lose them — refuse those too. (`XA COMMIT ... ONE + // PHASE`, though committed atomically, is likewise refused: at + // "XA START" time the one-phase outcome is unknowable.) The pending + // GTID is deliberately left unpromoted so the resume coordinate + // stays before the XA group. + if isXAStatement(q) { + return errXAUnsupported } ddlTables, err := extractTablesFromDDLStmts(string(event.Schema), string(event.Query)) if err != nil { @@ -715,7 +717,7 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { // unparseable statement on the server (e.g. a stored // procedure deploy in an unrelated schema) takes this path. c.promotePendingGTID() - return + return nil } // MySQL emits a synthetic GTID for DDL statements too, but the // DDL is its own transaction (no XIDEvent). Promote any pending @@ -727,6 +729,7 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { for _, ddlTable := range ddlTables { c.processDDLNotification(ddlTable.schema, ddlTable.table) } + return nil } // processTransactionPayload processes the events decompressed from a @@ -734,10 +737,10 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) { // per-session by any client regardless of the global value the preflight // checks). The payload carries the whole transaction except its GTID // event, so the pending {SID,GNO} stashed by the preceding (uncompressed) -// GTIDEvent is promoted here, by the inner group terminator — XIDEvent, -// COMMIT/ROLLBACK QueryEvent, or XA_PREPARE_LOG_EVENT — after the -// transaction's row events have been buffered, keeping -// promotePendingGTID's per-transaction resume contract intact. +// GTIDEvent is promoted here, by the inner group terminator — XIDEvent or +// COMMIT/ROLLBACK QueryEvent — after the transaction's row events have +// been buffered, keeping promotePendingGTID's per-transaction resume +// contract intact. func (c *gtidClient) processTransactionPayload(e *replication.TransactionPayloadEvent) error { for _, inner := range e.Events { switch innerEvent := inner.Event.(type) { @@ -749,7 +752,9 @@ func (c *gtidClient) processTransactionPayload(e *replication.TransactionPayload return err } case *replication.QueryEvent: - c.processQueryEvent(innerEvent) + if err := c.processQueryEvent(innerEvent); err != nil { + return err + } case *replication.TableMapEvent: // Already consumed by go-mysql's inner parser to decode the // RowsEvents above. @@ -758,13 +763,13 @@ func (c *gtidClient) processTransactionPayload(e *replication.TransactionPayload // 8.0.43): the first binlog group — XA START, row events, XA END, // XA_PREPARE_LOG_EVENT — arrives inside a payload, with only the // terminal XA COMMIT / XA ROLLBACK QueryEvent outside under its - // own GTID. The XA_PREPARE_LOG_EVENT (surfaced as a GenericEvent) - // terminates that first group, so promote exactly as readStream's - // GenericEvent case does — the group's row events have all been - // buffered by this point. + // own GTID. Spirit does not support XA workloads: the inner + // "XA START" QueryEvent above already fails the payload before + // any of its row events are buffered, so this branch — the + // XA_PREPARE_LOG_EVENT surfacing as a GenericEvent — is defense + // in depth, mirroring readStream's GenericEvent case. if inner.Header.EventType == replication.XA_PREPARE_LOG_EVENT { - c.promotePendingGTID() - continue + return errXAUnsupported } c.logger.Debug("Received unknown event type inside transaction payload", "type", inner.Header.EventType.String()) default: diff --git a/pkg/change/gtid_test.go b/pkg/change/gtid_test.go index f8539d78f..e600c923a 100644 --- a/pkg/change/gtid_test.go +++ b/pkg/change/gtid_test.go @@ -7,6 +7,7 @@ import ( "log/slog" "slices" "strings" + "sync/atomic" "testing" "time" @@ -195,9 +196,9 @@ func TestGTIDStartFromMalformedPosition(t *testing.T) { // TiDB parser cannot parse (CREATE TRIGGER, stored procedure bodies, // certain ALTER USER variants, ...) must still promote the // transaction's pending GTID into bufferedGTID. (XA statements are also -// unparseable but never reach the parser — they get explicit handling -// in readStream because promoting mid-XA-group would be incorrect; see -// TestGTIDClientXAPromotionOrdering.) Every QueryEvent on +// unparseable but never reach the parser — they fail the stream +// outright, since spirit does not support XA workloads; see +// TestGTIDClientXAGuardStream.) Every QueryEvent on // the entire server flows through the parser — the schema filter only // applies after parsing — so before the fix a single unparseable // statement in a *completely unrelated schema* left bufferedGTID @@ -373,19 +374,19 @@ func rollbackDanglingXATestTxns(t *testing.T, db *sql.DB, knownXIDs ...string) { } // TestGTIDClientXATransaction drives a real two-phase XA transaction -// (plus a one-phase variant) through the feed end-to-end. The binlog -// shape, verified against MySQL 8.0: nothing is written until XA -// PREPARE, at which point the entire first group — GTIDEvent(g1), -// Query("XA START ..."), the row events, Query("XA END ...") and the -// terminating XA_PREPARE_LOG_EVENT — is flushed at once, and g1 enters -// the server's gtid_executed. The terminal XA COMMIT (or XA ROLLBACK) -// arrives later as its own single-statement transaction under its own -// GTID (g2), with no row events. -// -// The BlockWait calls double as promotion-liveness assertions: BlockWait -// only returns once bufferedGTID covers the server's gtid_executed, so a -// handler that failed to promote at the XA prepare (or at the terminal -// XA COMMIT) would time out here. +// (plus a one-phase variant) against the feed end-to-end and asserts the +// XA guard fails the stream before any of the transaction's row events +// are buffered or applied. The binlog shape, verified against MySQL 8.0: +// nothing is written until XA PREPARE, at which point the entire first +// group — GTIDEvent(g1), Query("XA START ..."), the row events, +// Query("XA END ...") and the terminating XA_PREPARE_LOG_EVENT — is +// flushed at once. Spirit refuses the group at its opening "XA START" +// QueryEvent: prepare-time row images are written before the +// transaction's outcome is known, so applying them treats the prepare as +// a commit, and the XA ROLLBACK issued below would leave the target +// permanently diverged (nothing in the binlog undoes a rolled-back +// prepare). The abort is reported as a stream error so the caller +// preserves its checkpoint. // // Both the xids and the table names are unique per run. Unique xids // because XA START against a hard-coded xid fails with XAER_DUPID if an @@ -442,11 +443,25 @@ func TestGTIDClientXATransaction(t *testing.T) { cfg, err := mysql2.ParseDSN(testutils.DSN()) require.NoError(t, err) - client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), NewClientDefaultConfig()).(*gtidClient) - chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) - require.NoError(t, err) - require.NoError(t, client.AddSubscription(t1, t2, chunker)) - require.NoError(t, client.Start(t.Context())) + // newGuardClient builds and starts a client whose CancelFunc records + // the fatal reason, mimicking the runner's cancellation callback. + newGuardClient := func() (*gtidClient, *atomic.Int64) { + var gotReason atomic.Int64 + gotReason.Store(-1) + clientConfig := NewClientDefaultConfig() + clientConfig.CancelFunc = func(reason FatalReason) bool { + gotReason.Store(int64(reason)) + return true + } + client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), clientConfig).(*gtidClient) + chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) + require.NoError(t, err) + require.NoError(t, client.AddSubscription(t1, t2, chunker)) + require.NoError(t, client.Start(t.Context())) + return client, &gotReason + } + + client, gotReason := newGuardClient() defer client.Close() // XA transactions are session-scoped between XA START and XA @@ -466,73 +481,61 @@ func TestGTIDClientXATransaction(t *testing.T) { xaExec(fmt.Sprintf("XA END '%s'", xid)) // Before XA PREPARE nothing of the transaction exists in the binary - // log — no GTID has been assigned yet, so it cannot be in the - // buffered set, and no row events can have streamed. + // log — no GTID has been assigned yet, so no row events can have + // streamed and the guard cannot have fired. require.NoError(t, client.BlockWait(t.Context())) require.Equal(t, 0, client.GetDeltaLen(), "row events must not stream before XA PREPARE") + require.Equal(t, int64(-1), gotReason.Load(), "the guard must not fire before the XA group is binlogged") + // The prepare writes the whole group; the guard must fail the stream + // at its opening "XA START" QueryEvent — ahead of the row events — + // and classify it as a checkpoint-preserving stream error. xaExec(fmt.Sprintf("XA PREPARE '%s'", xid)) - - // The prepare flushes the whole group and terminates it. - require.NoError(t, client.BlockWait(t.Context())) - require.Equal(t, 2, client.GetDeltaLen(), "both row events must be buffered once XA PREPARE flushes the group") - - // The buffered images can flush before the XA COMMIT ever happens. - // (Known pre-existing property, shared with the binlog client: row - // images are applied from the prepare-time group, so a later XA - // ROLLBACK of the prepared transaction would not be compensated.) - require.NoError(t, client.Flush(t.Context())) + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, + 5*time.Second, 5*time.Millisecond, "XA PREPARE must fail the stream as a stream error") + client.streamWG.Wait() // reader fully exited: buffering is final + require.Equal(t, 0, client.GetDeltaLen(), "no prepared row events may be buffered once the guard fires") + + // Roll the prepared transaction back — the outcome spirit could never + // have seen coming. The target must not contain the prepared rows: + // applying them is exactly the permanent divergence the guard exists + // to prevent. + xaExec(fmt.Sprintf("XA ROLLBACK '%s'", xid)) var count int err = db.QueryRowContext(t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstTable)).Scan(&count) require.NoError(t, err) - require.Equal(t, 2, count) + require.Equal(t, 0, count, "prepared-then-rolled-back rows must never reach the target") + + // One-phase variant: `XA COMMIT ... ONE PHASE` commits atomically (a + // single group terminated by an XA_PREPARE_LOG_EVENT rather than a + // QueryEvent), but it is still an XA workload and the one-phase + // outcome is unknowable at the group's "XA START" — a fresh client + // must refuse it the same way. (Started after the rollback above, so + // the new stream — which begins at the current gtid_executed — only + // sees the one-phase group.) + client2, gotReason2 := newGuardClient() + defer client2.Close() - xaExec(fmt.Sprintf("XA COMMIT '%s'", xid)) - - // Capture gtid_executed right after the commit: it necessarily - // includes both the prepare-group GTID (g1) and the commit GTID (g2). - // Captured before BlockWait/Flush so the containment assertion below - // is deterministic even with unrelated concurrent load advancing - // gtid_executed on a shared server. - var executed string - require.NoError(t, db.QueryRowContext(t.Context(), "SELECT @@GLOBAL.gtid_executed").Scan(&executed)) - executedSet, err := mysql.ParseMysqlGTIDSet(normalizeGTIDString(executed)) - require.NoError(t, err) - - require.NoError(t, client.BlockWait(t.Context())) - require.NoError(t, client.Flush(t.Context())) - - // The resume coordinate must cover both XA GTIDs: resuming from it - // must not re-request (or worse, skip) any part of the XA transaction. - flushedSet, err := mysql.ParseMysqlGTIDSet(normalizeGTIDString(client.Position())) - require.NoError(t, err) - require.True(t, flushedSet.Contain(executedSet), - "flushed position %s must cover the executed set %s (both XA GTIDs)", flushedSet.String(), executedSet.String()) - - // One-phase variant: `XA COMMIT ... ONE PHASE` is a single group - // terminated by an XA_PREPARE_LOG_EVENT rather than a QueryEvent. xaExec(fmt.Sprintf("XA START '%s'", xid1p)) xaExec(fmt.Sprintf("INSERT INTO %s (a, b, c) VALUES (3, 4, 5)", srcTable)) xaExec(fmt.Sprintf("XA END '%s'", xid1p)) xaExec(fmt.Sprintf("XA COMMIT '%s' ONE PHASE", xid1p)) - require.NoError(t, client.BlockWait(t.Context())) - require.NoError(t, client.Flush(t.Context())) - err = db.QueryRowContext(t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstTable)).Scan(&count) - require.NoError(t, err) - require.Equal(t, 3, count) + require.Eventually(t, func() bool { return gotReason2.Load() == int64(FatalReasonStreamError) }, + 5*time.Second, 5*time.Millisecond, "one-phase XA must fail the stream too") + client2.streamWG.Wait() + require.Equal(t, 0, client2.GetDeltaLen(), "no one-phase XA row events may be buffered") } -// TestGTIDClientXATransactionCompression re-runs the XA two-phase dance -// with binlog_transaction_compression=ON on the XA session. The entire XA +// TestGTIDClientXATransactionCompression re-runs the XA guard dance with +// binlog_transaction_compression=ON on the XA session. The entire XA // first group — Query("XA START"), row events, Query("XA END") and the // terminating XA_PREPARE_LOG_EVENT — arrives inside one compressed // Transaction_payload event (verified against MySQL 8.0.43); only the -// terminal XA COMMIT QueryEvent stays outside, under its own GTID. The -// BlockWait after the prepare is the promotion-liveness assertion: it only -// returns if processTransactionPayload promoted the pending GTID at the -// inner XA_PREPARE_LOG_EVENT, and the delta count proves the inner row -// events were buffered rather than dropped. +// terminal XA COMMIT / XA ROLLBACK QueryEvent stays outside, under its +// own GTID. The inner "XA START" QueryEvent must fail the payload before +// the row events after it are buffered, surfacing exactly like the +// uncompressed abort: a checkpoint-preserving stream error. // // Unique xids and per-run table names for the same reasons documented on // TestGTIDClientXATransaction. @@ -542,7 +545,6 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { defer utils.CloseAndLog(db) xid := xaTestXIDPrefix + "_comp_" + uuid.NewString() - xid1p := xaTestXIDPrefix + "_comp1p_" + uuid.NewString() suffix := strings.ReplaceAll(uuid.NewString(), "-", "")[:8] srcTable := "gtidxacompt1_" + suffix dstTable := "gtidxacompt2_" + suffix @@ -561,7 +563,7 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { return } defer utils.CloseAndLog(cleanupDB) - rollbackDanglingXATestTxns(t, cleanupDB, xid, xid1p) + rollbackDanglingXATestTxns(t, cleanupDB, xid) _, _ = cleanupDB.ExecContext(context.Background(), "SET SESSION lock_wait_timeout=5") _, _ = cleanupDB.ExecContext(context.Background(), fmt.Sprintf("DROP TABLE IF EXISTS %s, %s", srcTable, dstTable)) }) @@ -574,7 +576,14 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { cfg, err := mysql2.ParseDSN(testutils.DSN()) require.NoError(t, err) - client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), NewClientDefaultConfig()).(*gtidClient) + var gotReason atomic.Int64 + gotReason.Store(-1) + clientConfig := NewClientDefaultConfig() + clientConfig.CancelFunc = func(reason FatalReason) bool { + gotReason.Store(int64(reason)) + return true + } + client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), clientConfig).(*gtidClient) chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) require.NoError(t, err) require.NoError(t, client.AddSubscription(t1, t2, chunker)) @@ -596,50 +605,39 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { xaExec(fmt.Sprintf("XA END '%s'", xid)) xaExec(fmt.Sprintf("XA PREPARE '%s'", xid)) - // The prepare flushes the whole group as one compressed payload. A - // handler that dropped the payload (or failed to promote at the inner - // XA_PREPARE_LOG_EVENT) would time out here. - require.NoError(t, client.BlockWait(t.Context())) - require.Equal(t, 2, client.GetDeltaLen(), "row events inside the compressed XA group must be buffered") - - xaExec(fmt.Sprintf("XA COMMIT '%s'", xid)) - require.NoError(t, client.BlockWait(t.Context())) - require.NoError(t, client.Flush(t.Context())) + // The prepare flushes the whole group as one compressed payload; the + // guard must fail it at the inner "XA START" QueryEvent, ahead of the + // inner row events. + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, + 5*time.Second, 5*time.Millisecond, "a compressed XA group must fail the stream as a stream error") + client.streamWG.Wait() // reader fully exited: buffering is final + require.Equal(t, 0, client.GetDeltaLen(), "no row events inside the compressed XA group may be buffered") + + // Roll the prepared transaction back and verify the target never saw + // the prepared rows — the divergence the guard exists to prevent. + xaExec(fmt.Sprintf("XA ROLLBACK '%s'", xid)) var count int require.NoError(t, db.QueryRowContext(t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstTable)).Scan(&count)) - require.Equal(t, 2, count) - - // One-phase variant: a single compressed group, likewise terminated by - // an XA_PREPARE_LOG_EVENT. - xaExec(fmt.Sprintf("XA START '%s'", xid1p)) - xaExec(fmt.Sprintf("INSERT INTO %s (a, b, c) VALUES (3, 4, 5)", srcTable)) - xaExec(fmt.Sprintf("XA END '%s'", xid1p)) - xaExec(fmt.Sprintf("XA COMMIT '%s' ONE PHASE", xid1p)) - - require.NoError(t, client.BlockWait(t.Context())) - require.NoError(t, client.Flush(t.Context())) - require.NoError(t, db.QueryRowContext(t.Context(), fmt.Sprintf("SELECT COUNT(*) FROM %s", dstTable)).Scan(&count)) - require.Equal(t, 3, count) + require.Equal(t, 0, count, "prepared-then-rolled-back rows must never reach the target") } -// TestGTIDClientXAPromotionOrdering is the deterministic regression test -// for the premature-promotion bug: the "XA START" QueryEvent used to -// fall through to the parser path (the TiDB parser cannot parse XA -// syntax) and promote the pending GTID before the transaction's row -// events had been buffered. A flush in that window published a resume -// coordinate that already covered the transaction, so a crash before -// the next flush resumed past it and silently lost its rows. +// TestGTIDClientXAGuardStream deterministically exercises the XA guard's +// readStream wiring: any XA event must fail the stream via +// CancelFunc(FatalReasonStreamError) — before the XA transaction's row +// events are buffered, and without promoting its GTID into the resume +// set (a resume must replay, and re-refuse, the XA group rather than +// skip it). // -// Events are injected through a synthetic go-mysql BinlogStreamer -// rather than a real server because the server writes an XA -// transaction's entire first group to the binlog in one burst at XA -// PREPARE time (see TestGTIDClientXATransaction): wall-clock timing -// cannot reliably observe the stream state between the "XA START" -// QueryEvent and the XA_PREPARE_LOG_EVENT of the same burst. Row events -// for a subscribed table are used as ordering barriers: events are -// consumed strictly in order, so once GetDeltaLen reflects a row event, -// every event injected before it has been processed. -func TestGTIDClientXAPromotionOrdering(t *testing.T) { +// Events are injected through a synthetic go-mysql BinlogStreamer rather +// than a real server because the server writes an XA transaction's +// entire first group to the binlog in one burst at XA PREPARE time (see +// TestGTIDClientXATransaction): wall-clock timing cannot reliably +// observe the stream state between the "XA START" QueryEvent and the +// row events of the same burst. Injection also reaches shapes a real +// 8.0 server never streams to us — a lone XA_PREPARE_LOG_EVENT without +// its opening "XA START" (the defense-in-depth branch), and a terminal +// XA COMMIT for a transaction prepared before we connected. +func TestGTIDClientXAGuardStream(t *testing.T) { db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) require.NoError(t, err) defer utils.CloseAndLog(db) @@ -657,23 +655,6 @@ func TestGTIDClientXAPromotionOrdering(t *testing.T) { cfg, err := mysql2.ParseDSN(testutils.DSN()) require.NoError(t, err) - client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), NewClientDefaultConfig()).(*gtidClient) - chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) - require.NoError(t, err) - require.NoError(t, client.AddSubscription(t1, t2, chunker)) - - // Wire readStream to a synthetic streamer instead of client.Start(). - empty, err := mysql.ParseMysqlGTIDSet("") - require.NoError(t, err) - streamer := replication.NewBinlogStreamer() - ctx, cancel := context.WithCancel(t.Context()) - client.streamer = streamer - client.bufferedGTID = empty - client.flushedGTID = empty.Clone() - client.cancelFunc = cancel - client.streamWG.Add(1) - go client.readStream(ctx) - defer client.Close() const sid = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee" sidUUID := uuid.MustParse(sid) @@ -698,63 +679,204 @@ func TestGTIDClientXAPromotionOrdering(t *testing.T) { }, } } - inject := func(evs ...*replication.BinlogEvent) { + xaPrepareEvent := func() *replication.BinlogEvent { + // go-mysql has no dedicated decoder for XA_PREPARE_LOG_EVENT, so + // it surfaces as a GenericEvent identified only by the header + // type — as in readStream itself. + return &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.XA_PREPARE_LOG_EVENT}, + Event: &replication.GenericEvent{}, + } + } + + // newSyntheticClient wires readStream to a synthetic streamer instead + // of client.Start(), with a CancelFunc that records the fatal reason. + newSyntheticClient := func(t *testing.T) (*gtidClient, *replication.BinlogStreamer, *atomic.Int64) { + var gotReason atomic.Int64 + gotReason.Store(-1) + clientConfig := NewClientDefaultConfig() + clientConfig.CancelFunc = func(reason FatalReason) bool { + gotReason.Store(int64(reason)) + return true + } + client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), clientConfig).(*gtidClient) + chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) + require.NoError(t, err) + require.NoError(t, client.AddSubscription(t1, t2, chunker)) + + empty, err := mysql.ParseMysqlGTIDSet("") + require.NoError(t, err) + streamer := replication.NewBinlogStreamer() + ctx, cancel := context.WithCancel(t.Context()) + client.streamer = streamer + client.bufferedGTID = empty + client.flushedGTID = empty.Clone() + client.cancelFunc = cancel + client.streamWG.Add(1) + go client.readStream(ctx) + t.Cleanup(client.Close) + return client, streamer, &gotReason + } + inject := func(t *testing.T, streamer *replication.BinlogStreamer, evs ...*replication.BinlogEvent) { t.Helper() for _, ev := range evs { require.NoError(t, streamer.AddEventToStreamer(ev)) } } - buffered := func(gno int64) bool { + // expectAbort waits for the guard to fire as a stream error and for + // readStream to fully exit, then verifies nothing was buffered: no + // row events, and no XA GTID in the resume set. + expectAbort := func(t *testing.T, client *gtidClient, gotReason *atomic.Int64, gno int64) { + t.Helper() + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, + 5*time.Second, 5*time.Millisecond, "the XA guard must fail the stream as a stream error") + client.streamWG.Wait() // reader fully exited: buffering is final + require.Equal(t, 0, client.GetDeltaLen(), "no row events may be buffered once the guard fires") target, err := mysql.ParseMysqlGTIDSet(fmt.Sprintf("%s:%d", sid, gno)) require.NoError(t, err) - return client.getBufferedGTID().Contain(target) + require.False(t, client.getBufferedGTID().Contain(target), + "the XA GTID must not enter the resume set: a resume must replay (and re-refuse) the XA group") } // The XA transaction's first group, exactly as the server writes it - // at XA PREPARE time. The row event doubles as the ordering barrier - // for the "XA START" QueryEvent before it. - inject(gtidEvent(100), queryEvent("XA START X'78',X'',1"), rowEvent(1)) - require.Eventually(t, func() bool { return client.GetDeltaLen() == 1 }, - 5*time.Second, 5*time.Millisecond, "row event after XA START was not processed") - require.False(t, buffered(100), - "the GTID must not be promoted at XA START: the transaction's row events are not buffered yet") - - // XA END does not terminate the group either. The second row event - // is the ordering barrier — the real group has none in this spot, - // but row events are inert to the promotion logic. - inject(queryEvent("XA END X'78',X'',1"), rowEvent(2)) - require.Eventually(t, func() bool { return client.GetDeltaLen() == 2 }, - 5*time.Second, 5*time.Millisecond, "row event after XA END was not processed") - require.False(t, buffered(100), - "the GTID must not be promoted at XA END: the group ends at the XA prepare event") - client.mu.Lock() - pendingGNO := client.pendingGNO - client.mu.Unlock() - require.EqualValues(t, 100, pendingGNO, "the XA transaction's GTID must still be pending after XA END") + // at XA PREPARE time. The guard must fire at the opening "XA START" + // QueryEvent; the row event injected after it must never be consumed. + t.Run("XA START", func(t *testing.T) { + client, streamer, gotReason := newSyntheticClient(t) + inject(t, streamer, gtidEvent(100), queryEvent("XA START X'78',X'',1"), rowEvent(1)) + expectAbort(t, client, gotReason, 100) + }) - // The XA_PREPARE_LOG_EVENT terminates the group. go-mysql has no - // dedicated decoder for it, so it surfaces as a GenericEvent - // identified only by the header type — as in readStream itself. - inject(&replication.BinlogEvent{ - Header: &replication.EventHeader{EventType: replication.XA_PREPARE_LOG_EVENT}, - Event: &replication.GenericEvent{}, + // A terminal XA COMMIT with no preceding "XA START" in-stream: the + // transaction was prepared before we connected, so its row events + // were never streamed and an applied commit would silently lose + // them. It must be refused, not promoted. Same for XA ROLLBACK. + t.Run("terminal XA COMMIT", func(t *testing.T) { + client, streamer, gotReason := newSyntheticClient(t) + inject(t, streamer, gtidEvent(101), queryEvent("XA COMMIT X'78',X'',1")) + expectAbort(t, client, gotReason, 101) + }) + t.Run("terminal XA ROLLBACK", func(t *testing.T) { + client, streamer, gotReason := newSyntheticClient(t) + inject(t, streamer, gtidEvent(102), queryEvent("XA ROLLBACK X'79',X'',1")) + expectAbort(t, client, gotReason, 102) + }) + + // The defense-in-depth branch: an XA_PREPARE_LOG_EVENT arriving + // without its group's "XA START" (no MySQL 8.0 server streams this + // shape today; the branch protects against a future regrouping). + t.Run("XA_PREPARE_LOG_EVENT", func(t *testing.T) { + client, streamer, gotReason := newSyntheticClient(t) + inject(t, streamer, gtidEvent(103), xaPrepareEvent()) + expectAbort(t, client, gotReason, 103) }) - require.Eventually(t, func() bool { return buffered(100) }, - 5*time.Second, 5*time.Millisecond, "the XA prepare event must promote the pending GTID") - - // The terminal XA COMMIT arrives later under its own GTID, with no - // row events. Same for an XA ROLLBACK outcome. - inject(gtidEvent(101), queryEvent("XA COMMIT X'78',X'',1")) - require.Eventually(t, func() bool { return buffered(101) }, - 5*time.Second, 5*time.Millisecond, "XA COMMIT must promote its own GTID") - - inject(gtidEvent(102), queryEvent("XA ROLLBACK X'79',X'',1")) - require.Eventually(t, func() bool { return buffered(102) }, - 5*time.Second, 5*time.Millisecond, "XA ROLLBACK must promote its own GTID") +} + +// TestGTIDProcessQueryEventXAGuard unit-tests processQueryEvent's +// statement classification directly: every XA statement the server +// binlogs as a QueryEvent must be refused with errXAUnsupported, while +// the non-XA transaction-control statements and DDL keep flowing +// through their existing nil-error paths (including statements that +// merely mention xa as an identifier). +func TestGTIDProcessQueryEventXAGuard(t *testing.T) { + empty, err := mysql.ParseMysqlGTIDSet("") + require.NoError(t, err) + c := >idClient{ + logger: slog.Default(), + subs: newSubscriptionRegistry(), + bufferedGTID: empty, + flushedGTID: empty.Clone(), + } + queryEvent := func(q string) *replication.QueryEvent { + return &replication.QueryEvent{Schema: []byte("test"), Query: []byte(q)} + } + // The canonical server-rewritten forms (XA BEGIN 'x' is binlogged + // with a hex-encoded xid), plus case and whitespace variations. + for _, q := range []string{ + "XA START X'78',X'',1", + "XA END X'78',X'',1", + "XA COMMIT X'78',X'',1", + "XA ROLLBACK X'78',X'',1", + "xa start X'78',X'',1", + " XA COMMIT X'78',X'',1 ", + } { + require.ErrorIs(t, c.processQueryEvent(queryEvent(q)), errXAUnsupported, "statement %q must be refused", q) + } + // Non-XA statements keep their existing behavior (no error). + for _, q := range []string{ + "BEGIN", + "COMMIT", + "ROLLBACK", + "SAVEPOINT `sp1`", + "ROLLBACK TO `sp1`", + "RELEASE SAVEPOINT `sp1`", + "CREATE TABLE xa_lookalike (a INT NOT NULL PRIMARY KEY)", + "DROP TABLE `xa`", // a table named xa is not an XA statement: the guard needs the keyword plus a space + } { + require.NoError(t, c.processQueryEvent(queryEvent(q)), "statement %q must not be refused", q) + } +} + +// TestGTIDProcessTransactionPayloadXAGuard unit-tests the compressed +// path directly: an inner "XA START" QueryEvent must fail the payload +// before the row events after it are buffered, and an inner +// XA_PREPARE_LOG_EVENT (the defense-in-depth branch) must fail it too. +func TestGTIDProcessTransactionPayloadXAGuard(t *testing.T) { + db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) + require.NoError(t, err) + defer utils.CloseAndLog(db) + + // Real tables so the subscription would buffer the inner row event + // if the guard failed to fire first. + testutils.RunSQL(t, "DROP TABLE IF EXISTS gtidxapayt1, gtidxapayt2") + testutils.RunSQL(t, "CREATE TABLE gtidxapayt1 (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))") + testutils.RunSQL(t, "CREATE TABLE gtidxapayt2 (a INT NOT NULL, b INT, c INT, PRIMARY KEY (a))") + + t1 := table.NewTableInfo(db, "test", "gtidxapayt1") + require.NoError(t, t1.SetInfo(t.Context())) + t2 := table.NewTableInfo(db, "test", "gtidxapayt2") + require.NoError(t, t2.SetInfo(t.Context())) + + cfg, err := mysql2.ParseDSN(testutils.DSN()) + require.NoError(t, err) + client := NewGTIDClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), NewClientDefaultConfig()).(*gtidClient) + chunker, err := table.NewChunker(t1, table.ChunkerConfig{NewTable: t2}) + require.NoError(t, err) + require.NoError(t, client.AddSubscription(t1, t2, chunker)) + defer client.Close() + + // A compressed XA prepare group as go-mysql decompresses it: the + // inner "XA START" QueryEvent precedes the inner row events. + xaGroup := &replication.TransactionPayloadEvent{Events: []*replication.BinlogEvent{ + { + Header: &replication.EventHeader{EventType: replication.QUERY_EVENT}, + Event: &replication.QueryEvent{Schema: []byte("test"), Query: []byte("XA START X'78',X'',1")}, + }, + { + Header: &replication.EventHeader{EventType: replication.WRITE_ROWS_EVENTv2}, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{Schema: []byte("test"), Table: []byte("gtidxapayt1")}, + Rows: [][]any{{int32(1), int32(0), int32(0)}}, + }, + }, + }} + require.ErrorIs(t, client.processTransactionPayload(xaGroup), errXAUnsupported) + require.Equal(t, 0, client.GetDeltaLen(), "the inner row events after XA START must not be buffered") + + // Defense in depth: an inner XA_PREPARE_LOG_EVENT without its + // opening "XA START". + prepareOnly := &replication.TransactionPayloadEvent{Events: []*replication.BinlogEvent{ + { + Header: &replication.EventHeader{EventType: replication.XA_PREPARE_LOG_EVENT}, + Event: &replication.GenericEvent{}, + }, + }} + require.ErrorIs(t, client.processTransactionPayload(prepareOnly), errXAUnsupported) + require.Equal(t, 0, client.GetDeltaLen()) } // TestGTIDClientSavepointPromotionOrdering is the savepoint twin of -// TestGTIDClientXAPromotionOrdering. MySQL logs "SAVEPOINT `sp1`" (and, in +// TestGTIDClientXAGuardStream. MySQL logs "SAVEPOINT `sp1`" (and, in // mixed-engine transactions, "ROLLBACK TO `sp1`") as a QueryEvent in the // *middle* of a row-format transaction group — verified against MySQL 8.0: // GTIDEvent → Query(BEGIN) → row events → Query("SAVEPOINT `sp1`") → more diff --git a/pkg/change/utils.go b/pkg/change/utils.go index bfe2219ce..f65b777af 100644 --- a/pkg/change/utils.go +++ b/pkg/change/utils.go @@ -3,6 +3,7 @@ package change import ( "context" "database/sql" + "errors" "fmt" "github.com/block/spirit/pkg/table" @@ -179,6 +180,32 @@ func checkImmutableColumn(tbl *table.TableInfo, ordinal int, beforeRow, afterRow return nil } +// errXAUnsupported is the fatal error produced when XA transaction +// activity is observed in the binlog stream. An XA transaction's row +// events are written to the binary log at XA PREPARE time, before the +// transaction's outcome is known: applying them treats the prepare as a +// commit, and a later XA ROLLBACK has no binlog representation that +// could undo them, so the target would diverge permanently. Rather than +// tracking prepared XIDs and buffering until the outcome (full XA +// support), spirit refuses XA workloads outright — the same posture as +// the preflight refusal of non-empty binlog_row_value_options. Both +// change clients treat this as a fatal stream error, aborting before +// any of the XA transaction's row events are buffered. +var errXAUnsupported = errors.New("XA transactions detected in the binlog stream: spirit does not support XA workloads") + +// isXAStatement reports whether q (a binlogged statement, whitespace +// already trimmed) is one of the XA transaction statements MySQL writes +// to the binary log as QueryEvents: "XA START", "XA END", "XA COMMIT" +// or "XA ROLLBACK". (Two-phase "XA PREPARE" is logged as an +// XA_PREPARE_LOG_EVENT, not a QueryEvent.) The server rewrites XA +// statements canonically before logging — XA BEGIN 'x' is binlogged as +// "XA START" with a hex-encoded xid — so a keyword prefix match is +// exact, and no other statement the server binlogs begins with the XA +// keyword. +func isXAStatement(q string) bool { + return hasPrefixFold(q, "XA ") +} + // isMinimalRowImage returns true if the RowsEvent contains a minimal row image, // i.e. some columns were skipped. This happens when binlog_row_image=MINIMAL or NOBLOB. // With full row images, SkippedColumns entries are empty slices. From 8b08a53cd692a7450e98393f849a93f2ead39f0f Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 24 Sep 2026 08:54:45 -0600 Subject: [PATCH 2/4] Run XA tests against an isolated MySQL CI server --- .github/workflows/mysql-xa-docker.yml | 34 ++++++++++++++++++++++ compose/compose.yml | 16 +++++++++- compose/replication-tls/replication-ci.yml | 3 +- 3 files changed, 51 insertions(+), 2 deletions(-) create mode 100644 .github/workflows/mysql-xa-docker.yml diff --git a/.github/workflows/mysql-xa-docker.yml b/.github/workflows/mysql-xa-docker.yml new file mode 100644 index 000000000..28698a11c --- /dev/null +++ b/.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." diff --git a/compose/compose.yml b/compose/compose.yml index 8dea7d8c7..d10d3fe7c 100644 --- a/compose/compose.yml +++ b/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 diff --git a/compose/replication-tls/replication-ci.yml b/compose/replication-tls/replication-ci.yml index 101eb35b8..22a8b4a92 100644 --- a/compose/replication-tls/replication-ci.yml +++ b/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 From b7d9303fcad80be6d4839d16abe91acb6376e9d3 Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 24 Sep 2026 09:05:35 -0600 Subject: [PATCH 3/4] Classify binlog query events with SQL parser --- pkg/change/binlog.go | 39 ++++++------- pkg/change/gtid.go | 119 ++++++++------------------------------- pkg/change/gtid_test.go | 2 +- pkg/change/utils.go | 77 ++++++++++++++++--------- pkg/change/utils_test.go | 26 +++++++++ 5 files changed, 115 insertions(+), 148 deletions(-) diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go index 8a2efce45..07653e2a1 100644 --- a/pkg/change/binlog.go +++ b/pkg/change/binlog.go @@ -774,6 +774,13 @@ func (c *binlogClient) readStream(ctx context.Context) { return } case *replication.QueryEvent: + info, err := parseQueryEvent(string(event.Schema), string(event.Query)) + if err != nil { + // 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 + } // 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 @@ -783,26 +790,14 @@ func (c *binlogClient) readStream(ctx context.Context) { // 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 isXAStatement(strings.TrimSpace(string(event.Query))) { + if info.xa { c.logger.Error("fatal error processing binlog query event", "error", errXAUnsupported) c.fatalError(FatalReasonStreamError) return } // 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)) - 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. - c.logger.Error("Skipping query that was unable to parse", "file", currentLogName, "pos", ev.Header.LogPos) - continue - } - for _, ddlTable := range ddlTables { + for _, ddlTable := range info.tables { c.processDDLNotification(ddlTable.schema, ddlTable.table) } case *replication.TransactionPayloadEvent: @@ -1067,23 +1062,23 @@ func (c *binlogClient) processTransactionPayload(e *replication.TransactionPaylo return err } case *replication.QueryEvent: + 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 + } // 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 isXAStatement(strings.TrimSpace(string(innerEvent.Query))) { + 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. - ddlTables, _, err := extractTablesFromDDLStmts(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 { + for _, ddlTable := range info.tables { c.processDDLNotification(ddlTable.schema, ddlTable.table) } case *replication.TableMapEvent, *replication.XIDEvent: diff --git a/pkg/change/gtid.go b/pkg/change/gtid.go index d3942f2ea..7c5560ec2 100644 --- a/pkg/change/gtid.go +++ b/pkg/change/gtid.go @@ -279,12 +279,6 @@ func normalizeGTIDString(s string) string { }, s) } -// hasPrefixFold reports whether s begins with prefix, matched -// case-insensitively (strings.HasPrefix + strings.EqualFold). -func hasPrefixFold(s, prefix string) bool { - return len(s) >= len(prefix) && strings.EqualFold(s[:len(prefix)], prefix) -} - // setBufferedGTID adds gtid to bufferedGTID under c.mu. The set grows // monotonically — there is no "rewind" in GTID semantics. func (c *gtidClient) setBufferedGTID(sid []byte, gno int64) { @@ -749,89 +743,13 @@ func (c *gtidClient) readStream(ctx context.Context) { // processQueryEvent handles a QueryEvent, whether read directly from the // stream or decompressed from a transaction payload. Transaction-control -// statements adjust the pending-GTID state, XA statements fail the stream -// (spirit does not support XA workloads; see the guard below), and -// everything else goes through DDL extraction. See promotePendingGTID for +// statements adjust the pending-GTID state, XA statements fail the stream, +// and DDL statements notify subscribers. See promotePendingGTID for // the group shapes that dictate which statements promote and which must // leave the pending GTID pending. A returned error is fatal: the caller // must tear the stream down without buffering anything further. func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { - // A "BEGIN" QueryEvent inside a transaction is not DDL — skip - // it cheaply rather than handing it to the parser. The pending - // GTID must stay pending: the transaction's row events have not - // been buffered yet. - q := strings.TrimSpace(string(event.Query)) - if strings.EqualFold(q, "BEGIN") { - return nil - } - // MySQL also logs SAVEPOINT statements as QueryEvents in the - // *middle* of a row-format transaction (verified against MySQL - // 8.0: GTIDEvent → Query(BEGIN) → row events → - // Query("SAVEPOINT `sp1`") → more row events → XIDEvent). - // "ROLLBACK TO `sp1`" appears mid-group the same way when - // non-transactional writes after the savepoint prevent the - // server from simply truncating its binlog cache. Neither - // terminates the group — it still ends at the XIDEvent or - // COMMIT/ROLLBACK QueryEvent that follows — so, exactly like - // BEGIN and "XA START"/"XA END", the pending GTID must stay - // pending: falling through to the parser path below would - // promote on both of its branches, letting a concurrent flush - // publish the GTID as a resume coordinate before the rest of - // the transaction's row events are buffered. A crash before the - // next flush would then resume past the transaction and - // silently lose its tail. The server rewrites these statements - // with backtick-quoted identifiers ("ROLLBACK TO `sp1`" — no - // SAVEPOINT keyword), so keyword prefix matches are exact. - // "ROLLBACK TO " is matched here, above the terminator check - // below, so it can never be taken for a transaction-ending - // ROLLBACK; "RELEASE SAVEPOINT " is matched defensively (MySQL - // does not binlog it today) since releasing a savepoint never - // ends a transaction either. - if hasPrefixFold(q, "SAVEPOINT ") || hasPrefixFold(q, "ROLLBACK TO ") || hasPrefixFold(q, "RELEASE SAVEPOINT ") { - return nil - } - // COMMIT/ROLLBACK QueryEvents end a transaction that involved a - // non-transactional engine (these get a QueryEvent terminator - // instead of an XIDEvent; a logged ROLLBACK is the mixed-engine - // case where the non-transactional writes survived the rollback). - // Either way the server has recorded the GTID in gtid_executed - // and we have buffered all of the transaction's row events, so - // promote — exactly as the XIDEvent path does. Skipping the - // promotion here would wedge BlockWait forever. - if strings.EqualFold(q, "COMMIT") || strings.EqualFold(q, "ROLLBACK") { - c.promotePendingGTID() - return nil - } - // Any XA statement fails the stream: spirit does not support XA - // workloads. An XA transaction's first binlog group is written in - // one piece at XA PREPARE time (verified against MySQL 8.0): - // - // GTIDEvent(g1) → Query("XA START x") → row events → - // Query("XA END x") → XA_PREPARE_LOG_EVENT - // - // with the terminal XA COMMIT or XA ROLLBACK arriving any amount - // of time later as a QueryEvent under its own GTID (g2), with no - // row events. The row events are therefore streamed before the - // transaction's outcome is known: buffering and flushing them - // treats the prepare as a commit, and a later XA ROLLBACK has no - // binlog representation that could undo them — the target would - // diverge permanently, detectable only by checksum. Rather than - // track prepared XIDs and buffer until the outcome, refuse the - // workload. Failing on "XA START" — before any of the group's row - // events — guarantees none of them are ever buffered, let alone - // flushed. A terminal XA COMMIT / XA ROLLBACK with no preceding - // "XA START" in-stream means the transaction was prepared before - // we connected: its row events were never streamed and may postdate - // the copier's snapshot of their chunk, so an XA COMMIT outcome - // could silently lose them — refuse those too. (`XA COMMIT ... ONE - // PHASE`, though committed atomically, is likewise refused: at - // "XA START" time the one-phase outcome is unknowable.) The pending - // GTID is deliberately left unpromoted so the resume coordinate - // stays before the XA group. - if isXAStatement(q) { - return errXAUnsupported - } - ddlTables, opensTransaction, 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 (mode-dependent SQL // such as ANSI_QUOTES quoting, or syntax newer than the grammar) @@ -871,6 +789,23 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { // path. return nil } + // A parsed XA statement is unsafe even when it is a terminal COMMIT or + // ROLLBACK from a transaction prepared before this stream connected. + // The XA prepare group can expose row events before the outcome is known. + if info.xa { + return errXAUnsupported + } + // BEGIN, SAVEPOINT, and ROLLBACK TO SAVEPOINT leave the group open. + // Its row events must be buffered before the pending GTID is promoted. + if info.opensTransaction || info.keepsTransactionOpen { + return nil + } + // Mixed-engine transactions use a COMMIT/ROLLBACK QueryEvent in place + // of XIDEvent. The transaction is complete at this point. + if info.endsTransaction { + c.promotePendingGTID() + return nil + } // MySQL emits a synthetic GTID for DDL statements too, but the // DDL is its own transaction (no XIDEvent). Promote any pending // GTID now so a DDL-as-last-event still ends up in the resume @@ -878,18 +813,8 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { // won't actually resume, but the position is consistent for // non-cancelling filters. // - // The exception is a statement that *opens* its group: the parsed - // CREATE TABLE ... START TRANSACTION form of CTAS (its row events - // are still to come, exactly like BEGIN above — the XIDEvent that - // ends the group promotes), or a hypothetical spelled-out - // START TRANSACTION the BEGIN fast-path above didn't catch. - // Promoting here would let a concurrent flush publish the GTID as - // a resume coordinate before the group's row events are buffered, - // silently losing them on resume. - if !opensTransaction { - c.promotePendingGTID() - } - for _, ddlTable := range ddlTables { + c.promotePendingGTID() + for _, ddlTable := range info.tables { c.processDDLNotification(ddlTable.schema, ddlTable.table) } return nil diff --git a/pkg/change/gtid_test.go b/pkg/change/gtid_test.go index c087d8e3b..0e5162001 100644 --- a/pkg/change/gtid_test.go +++ b/pkg/change/gtid_test.go @@ -1061,7 +1061,7 @@ func TestGTIDProcessQueryEventXAGuard(t *testing.T) { "ROLLBACK TO `sp1`", "RELEASE SAVEPOINT `sp1`", "CREATE TABLE xa_lookalike (a INT NOT NULL PRIMARY KEY)", - "DROP TABLE `xa`", // a table named xa is not an XA statement: the guard needs the keyword plus a space + "DROP TABLE `xa`", // a table named xa is not an XA statement } { require.NoError(t, c.processQueryEvent(queryEvent(q)), "statement %q must not be refused", q) } diff --git a/pkg/change/utils.go b/pkg/change/utils.go index 44094dda5..0d47404ca 100644 --- a/pkg/change/utils.go +++ b/pkg/change/utils.go @@ -42,35 +42,48 @@ type schemaTable struct { table string } -// extractTablesFromDDLStmts extracts table names from DDL statements. -// The logic is based on canal: https://github.com/go-mysql-org/go-mysql/blob/34b6b0998dde44e51dff0bbcc1ac88339f57f830/canal/sync.go#L195-L245 -// -// opensTransaction reports that the statement opens a transaction group -// rather than being one: BEGIN / START TRANSACTION, or the -// CREATE TABLE ... START TRANSACTION form MySQL 8.0.21+ writes to the -// binary log in place of CREATE TABLE ... SELECT under row-based -// replication. The group's row events follow the statement, so GTID -// promotion must wait for the group's real terminator (see -// gtidClient.processQueryEvent). -func extractTablesFromDDLStmts(defaultSchema string, statements string) (tables []schemaTable, opensTransaction bool, err error) { +// queryEventInfo describes the statements in one binlog QueryEvent. +type queryEventInfo struct { + tables []schemaTable + opensTransaction bool + keepsTransactionOpen bool + endsTransaction bool + xa bool +} + +// parseQueryEvent classifies transaction control and extracts DDL table names +// from the same parse, so every consumer sees the same statement semantics. +func parseQueryEvent(defaultSchema, statements string) (info queryEventInfo, err error) { p := parser.New() stmts, _, err := p.Parse(statements, "", "") if err != nil { - return nil, false, err + return queryEventInfo{}, err } for _, stmt := range stmts { switch t := stmt.(type) { + case *ast.XAStmt: + info.xa = true case *ast.BeginStmt: - opensTransaction = true + info.opensTransaction = true + case *ast.SavepointStmt, *ast.ReleaseSavepointStmt: + info.keepsTransactionOpen = true + case *ast.RollbackStmt: + if t.SavepointName != "" { + info.keepsTransactionOpen = true + } else { + info.endsTransaction = true + } + case *ast.CommitStmt: + info.endsTransaction = true case *ast.RenameTableStmt: for _, tableInfo := range t.TableToTables { schema, table := getTableIdentity(defaultSchema, tableInfo.OldTable) - tables = append(tables, schemaTable{schema, table}) + info.tables = append(info.tables, schemaTable{schema, table}) } case *ast.DropTableStmt: for _, table := range t.Tables { schema, tableName := getTableIdentity(defaultSchema, table) - tables = append(tables, schemaTable{schema, tableName}) + info.tables = append(info.tables, schemaTable{schema, tableName}) } case *ast.AlterTableStmt, *ast.CreateTableStmt, *ast.TruncateTableStmt, *ast.CreateIndexStmt, *ast.DropIndexStmt: @@ -81,7 +94,7 @@ func extractTablesFromDDLStmts(defaultSchema string, statements string) (tables case *ast.CreateTableStmt: tableNode = n.Table if n.StartTransaction { - opensTransaction = true + info.opensTransaction = true } case *ast.TruncateTableStmt: tableNode = n.Table @@ -91,10 +104,28 @@ func extractTablesFromDDLStmts(defaultSchema string, statements string) (tables tableNode = n.Table } schema, table := getTableIdentity(defaultSchema, tableNode) - tables = append(tables, schemaTable{schema, table}) + info.tables = append(info.tables, schemaTable{schema, table}) } } - return tables, opensTransaction, nil + return info, nil +} + +// extractTablesFromDDLStmts extracts table names from DDL statements. +// The logic is based on canal: https://github.com/go-mysql-org/go-mysql/blob/34b6b0998dde44e51dff0bbcc1ac88339f57f830/canal/sync.go#L195-L245 +// +// opensTransaction reports that the statement opens a transaction group +// rather than being one: BEGIN / START TRANSACTION, or the +// CREATE TABLE ... START TRANSACTION form MySQL 8.0.21+ writes to the +// binary log in place of CREATE TABLE ... SELECT under row-based +// replication. The group's row events follow the statement, so GTID +// promotion must wait for the group's real terminator (see +// gtidClient.processQueryEvent). +func extractTablesFromDDLStmts(defaultSchema string, statements string) (tables []schemaTable, opensTransaction bool, err error) { + info, err := parseQueryEvent(defaultSchema, statements) + if err != nil { + return nil, false, err + } + return info.tables, info.opensTransaction, nil } // toSet converts a string slice to a set (map[string]struct{}) for O(1) lookups. @@ -204,16 +235,6 @@ func checkImmutableColumn(tbl *table.TableInfo, ordinal int, beforeRow, afterRow // any of the XA transaction's row events are buffered. var errXAUnsupported = errors.New("XA transactions detected in the binlog stream: spirit does not support XA workloads") -// isXAStatement reports whether q (a binlogged statement, whitespace -// already trimmed) begins with the XA keyword. MySQL normally writes -// "XA START", "XA END", "XA COMMIT", and "XA ROLLBACK" as QueryEvents; -// two-phase "XA PREPARE" is logged as an XA_PREPARE_LOG_EVENT. Matching -// any XA statement also refuses new QueryEvent forms safely. The server -// rewrites XA BEGIN 'x' as "XA START" with a hex-encoded xid. -func isXAStatement(q string) bool { - return hasPrefixFold(q, "XA ") -} - // isMinimalRowImage returns true if the RowsEvent contains a minimal row image, // i.e. some columns were skipped. This happens when binlog_row_image=MINIMAL or NOBLOB. // With full row images, SkippedColumns entries are empty slices. diff --git a/pkg/change/utils_test.go b/pkg/change/utils_test.go index 988aeb007..76325a5be 100644 --- a/pkg/change/utils_test.go +++ b/pkg/change/utils_test.go @@ -7,6 +7,32 @@ import ( "github.com/stretchr/testify/require" ) +func TestParseQueryEventClassification(t *testing.T) { + tests := []struct { + query string + want queryEventInfo + }{ + {" xa begin 'x' ", queryEventInfo{xa: true}}, + {"XA COMMIT X'78' ONE PHASE", queryEventInfo{xa: true}}, + {"XA ROLLBACK X'78'", queryEventInfo{xa: true}}, + {"BEGIN", queryEventInfo{opensTransaction: true}}, + {"START TRANSACTION", queryEventInfo{opensTransaction: true}}, + {"SAVEPOINT `s`", queryEventInfo{keepsTransactionOpen: true}}, + {"ROLLBACK TO `s`", queryEventInfo{keepsTransactionOpen: true}}, + {"RELEASE SAVEPOINT `s`", queryEventInfo{keepsTransactionOpen: true}}, + {"COMMIT", queryEventInfo{endsTransaction: true}}, + {"ROLLBACK", queryEventInfo{endsTransaction: true}}, + {"CREATE TABLE xa_lookalike (id INT PRIMARY KEY)", queryEventInfo{tables: []schemaTable{{"test", "xa_lookalike"}}}}, + } + for _, tt := range tests { + t.Run(tt.query, func(t *testing.T) { + got, err := parseQueryEvent("test", tt.query) + require.NoError(t, err) + require.Equal(t, tt.want, got) + }) + } +} + func TestEncodeSchemaTable(t *testing.T) { tests := []struct { name string From 3fe6cc4ca56406e30ce2149b62bee7d79b29e6e7 Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 24 Sep 2026 10:03:31 -0600 Subject: [PATCH 4/4] Preserve CTAS notifications and classify XA aborts --- pkg/change/binlog.go | 6 +-- pkg/change/binlog_test.go | 41 ++++++++++++++++++-- pkg/change/config.go | 9 ++++- pkg/change/gtid.go | 27 +++++++------- pkg/change/gtid_test.go | 60 ++++++++++++++++++++---------- pkg/change/utils.go | 18 ++++++++- pkg/datasync/runner.go | 3 +- pkg/datasync/sync_test.go | 7 ++++ pkg/migration/runner.go | 3 ++ pkg/migration/runner_close_test.go | 13 +++++++ pkg/move/runner.go | 3 ++ pkg/move/runner_close_test.go | 9 +++++ 12 files changed, 156 insertions(+), 43 deletions(-) diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go index 07653e2a1..b36f46f9c 100644 --- a/pkg/change/binlog.go +++ b/pkg/change/binlog.go @@ -792,7 +792,7 @@ func (c *binlogClient) readStream(ctx context.Context) { // for the full rationale and group shape. if info.xa { c.logger.Error("fatal error processing binlog query event", "error", errXAUnsupported) - c.fatalError(FatalReasonStreamError) + c.fatalError(FatalReasonUnsupportedXA) return } // Query event, check if it is a DDL statement, @@ -826,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, @@ -852,7 +852,7 @@ func (c *binlogClient) readStream(ctx context.Context) { // 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(FatalReasonStreamError) + c.fatalError(FatalReasonUnsupportedXA) return } c.logger.Debug("Received unknown event type", "type", ev.Header.EventType.String()) diff --git a/pkg/change/binlog_test.go b/pkg/change/binlog_test.go index 34a1f6600..922455c81 100644 --- a/pkg/change/binlog_test.go +++ b/pkg/change/binlog_test.go @@ -702,8 +702,8 @@ func TestDDLNotificationTransactionCompression(t *testing.T) { // terminating XA_PREPARE_LOG_EVENT — is written to the binlog in one // burst at XA PREPARE time; the guard must fail the stream at the // opening "XA START" QueryEvent, before any of the row events after it -// are buffered, and classify the abort as a checkpoint-preserving -// stream error. Unique xids and per-run table names for the reasons +// are buffered, and classify the abort as an unsupported-XA +// reason. Unique xids and per-run table names for the reasons // documented on TestGTIDClientXATransaction. func TestBinlogClientXATransactionGuard(t *testing.T) { db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) @@ -780,8 +780,8 @@ func TestBinlogClientXATransactionGuard(t *testing.T) { require.Equal(t, int64(-1), gotReason.Load(), "the guard must not fire before the XA group is binlogged") xaExec(fmt.Sprintf("XA PREPARE '%s'", xid)) - require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, - 5*time.Second, 5*time.Millisecond, "XA PREPARE must fail the stream as a stream error") + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonUnsupportedXA) }, + 5*time.Second, 5*time.Millisecond, "XA PREPARE must report unsupported XA") client.streamWG.Wait() // reader fully exited: buffering is final require.Equal(t, 0, client.GetDeltaLen(), "no prepared row events may be buffered once the guard fires") @@ -852,6 +852,39 @@ func TestBinlogProcessTransactionPayloadXAGuard(t *testing.T) { require.Equal(t, 0, client.GetDeltaLen()) } +// A lone XA_PREPARE_LOG_EVENT must trigger the uncompressed stream guard, +// even if a future server version omits the opening XA QueryEvent. +func TestBinlogClientXAPrepareEventGuard(t *testing.T) { + db, err := dbconn.New(testutils.DSN(), dbconn.NewDBConfig()) + require.NoError(t, err) + defer utils.CloseAndLog(db) + cfg, err := mysql2.ParseDSN(testutils.DSN()) + require.NoError(t, err) + + var gotReason atomic.Int64 + gotReason.Store(-1) + clientConfig := NewClientDefaultConfig() + clientConfig.CancelFunc = func(reason FatalReason) bool { + gotReason.Store(int64(reason)) + return true + } + client := NewBinlogClient(db, cfg.Addr, cfg.User, cfg.Passwd, applier.NewSingleTargetForTest(t, db), clientConfig).(*binlogClient) + streamer := replication.NewBinlogStreamer() + ctx, cancel := context.WithCancel(t.Context()) + client.streamer = streamer + client.cancelFunc = cancel + client.streamWG.Add(1) + go client.readStream(ctx) + defer client.Close() + + require.NoError(t, streamer.AddEventToStreamer(&replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: replication.XA_PREPARE_LOG_EVENT}, + Event: &replication.GenericEvent{}, + })) + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonUnsupportedXA) }, + 5*time.Second, 5*time.Millisecond, "a lone XA_PREPARE_LOG_EVENT must report unsupported XA") +} + // TestCompositePKUpdate tests that we correctly handle // the case when a PRIMARY KEY is moved. // See: https://github.com/block/spirit/issues/417 diff --git a/pkg/change/config.go b/pkg/change/config.go index 5f23430c7..7b493bd14 100644 --- a/pkg/change/config.go +++ b/pkg/change/config.go @@ -23,6 +23,10 @@ const ( // processed). The watched tables are not known to have changed, so // persisted resume state remains valid and a retry can resume from it. FatalReasonStreamError + // FatalReasonUnsupportedXA means an XA group was found in the stream. + // Replaying the checkpoint would hit the same group, so finite runs + // must discard their checkpoint and start fresh after XA activity stops. + FatalReasonUnsupportedXA ) // String implements fmt.Stringer for logging. @@ -32,6 +36,8 @@ func (f FatalReason) String() string { return "schema-change" case FatalReasonStreamError: return "stream-error" + case FatalReasonUnsupportedXA: + return "unsupported-xa" default: return fmt.Sprintf("unknown-fatal-reason(%d)", int(f)) } @@ -46,7 +52,8 @@ type ClientConfig struct { // It is called when a DDL change is detected on a subscribed table // (FatalReasonSchemaChange), or when a fatal stream error occurs, such as // minimal RBR detection or exhausted streamer recreation attempts - // (FatalReasonStreamError). The caller is expected to handle cancellation + // (FatalReasonStreamError), or when XA is detected + // (FatalReasonUnsupportedXA). The caller is expected to handle cancellation // and cleanup, using reason to decide whether persisted resume state // (e.g. a checkpoint) must be invalidated (schema change) or is still // safe to resume from (stream error). diff --git a/pkg/change/gtid.go b/pkg/change/gtid.go index 7c5560ec2..92f0fdbe9 100644 --- a/pkg/change/gtid.go +++ b/pkg/change/gtid.go @@ -690,7 +690,7 @@ func (c *gtidClient) readStream(ctx context.Context) { case *replication.QueryEvent: if err = c.processQueryEvent(event); err != nil { c.logger.Error("fatal error processing GTID query event", "error", err) - c.fatalError(FatalReasonStreamError) + c.fatalError(fatalReasonForStreamError(err)) return } case *replication.TransactionPayloadEvent: @@ -705,7 +705,7 @@ func (c *gtidClient) readStream(ctx context.Context) { // wedging BlockWait/Flush forever. if err = c.processTransactionPayload(event); err != nil { c.logger.Error("fatal error processing GTID transaction payload event", "error", err) - c.fatalError(FatalReasonStreamError) + c.fatalError(fatalReasonForStreamError(err)) return } case *replication.RotateEvent: @@ -731,7 +731,7 @@ func (c *gtidClient) readStream(ctx context.Context) { // version reshapes the group. if ev.Header.EventType == replication.XA_PREPARE_LOG_EVENT { c.logger.Error("fatal error processing GTID stream", "error", errXAUnsupported) - c.fatalError(FatalReasonStreamError) + c.fatalError(FatalReasonUnsupportedXA) return } c.logger.Debug("Received unknown event type", "type", ev.Header.EventType.String()) @@ -795,9 +795,9 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { if info.xa { return errXAUnsupported } - // BEGIN, SAVEPOINT, and ROLLBACK TO SAVEPOINT leave the group open. - // Its row events must be buffered before the pending GTID is promoted. - if info.opensTransaction || info.keepsTransactionOpen { + // SAVEPOINT and ROLLBACK TO SAVEPOINT leave the group open without + // naming a DDL table. Its row events must precede GTID promotion. + if info.keepsTransactionOpen { return nil } // Mixed-engine transactions use a COMMIT/ROLLBACK QueryEvent in place @@ -806,14 +806,13 @@ func (c *gtidClient) processQueryEvent(event *replication.QueryEvent) error { c.promotePendingGTID() return nil } - // MySQL emits a synthetic GTID for DDL statements too, but the - // DDL is its own transaction (no XIDEvent). Promote any pending - // GTID now so a DDL-as-last-event still ends up in the resume - // set. This is best-effort — if the caller cancels on DDL we - // won't actually resume, but the position is consistent for - // non-cancelling filters. - // - c.promotePendingGTID() + // Ordinary DDL is its own transaction. A statement that opens a group, + // such as BEGIN or CREATE TABLE ... START TRANSACTION, must wait for its + // terminator before promotion. The CREATE TABLE still needs to notify + // DDL subscribers now, even though its GTID remains pending. + if !info.opensTransaction { + c.promotePendingGTID() + } for _, ddlTable := range info.tables { c.processDDLNotification(ddlTable.schema, ddlTable.table) } diff --git a/pkg/change/gtid_test.go b/pkg/change/gtid_test.go index 0e5162001..8223319c9 100644 --- a/pkg/change/gtid_test.go +++ b/pkg/change/gtid_test.go @@ -424,9 +424,8 @@ func TestGTIDResumeAfterGTIDHistoryRegression(t *testing.T) { // pending GTID into bufferedGTID — not at the QueryEvent itself (it // could sit mid-group; see // TestGTIDClientQueryPromotionOrdering), but at the next -// GTIDEvent, which proves the group ended. (XA statements are also -// unparseable but never reach the parser — they fail the stream; see -// TestGTIDClientXAGuardStream.) Every QueryEvent on +// GTIDEvent, which proves the group ended. XA statements parse as +// *ast.XAStmt and fail the stream; see TestGTIDClientXAGuardStream. Every QueryEvent on // the entire server flows through the parser — the schema filter only // applies after parsing — so before the fix a single unparseable // statement in a *completely unrelated schema* left bufferedGTID @@ -633,8 +632,8 @@ func rollbackDanglingXATestTxns(t *testing.T, db *sql.DB, knownXIDs ...string) { // transaction's outcome is known, so applying them treats the prepare as // a commit, and the XA ROLLBACK issued below would leave the target // permanently diverged (nothing in the binlog undoes a rolled-back -// prepare). The abort is reported as a stream error so the caller -// preserves its checkpoint. +// prepare). The abort is reported as an unsupported-XA reason so the caller +// invalidates its checkpoint and requires a fresh start. // // Both the xids and the table names are unique per run. Unique xids // because XA START against a hard-coded xid fails with XAER_DUPID if an @@ -738,10 +737,10 @@ func TestGTIDClientXATransaction(t *testing.T) { // The prepare writes the whole group; the guard must fail the stream // at its opening "XA START" QueryEvent — ahead of the row events — - // and classify it as a checkpoint-preserving stream error. + // and classify it as an unsupported-XA reason. xaExec(fmt.Sprintf("XA PREPARE '%s'", xid)) - require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, - 5*time.Second, 5*time.Millisecond, "XA PREPARE must fail the stream as a stream error") + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonUnsupportedXA) }, + 5*time.Second, 5*time.Millisecond, "XA PREPARE must fail the stream as unsupported XA") client.streamWG.Wait() // reader fully exited: buffering is final require.Equal(t, 0, client.GetDeltaLen(), "no prepared row events may be buffered once the guard fires") @@ -770,7 +769,7 @@ func TestGTIDClientXATransaction(t *testing.T) { xaExec(fmt.Sprintf("XA END '%s'", xid1p)) xaExec(fmt.Sprintf("XA COMMIT '%s' ONE PHASE", xid1p)) - require.Eventually(t, func() bool { return gotReason2.Load() == int64(FatalReasonStreamError) }, + require.Eventually(t, func() bool { return gotReason2.Load() == int64(FatalReasonUnsupportedXA) }, 5*time.Second, 5*time.Millisecond, "one-phase XA must fail the stream too") client2.streamWG.Wait() require.Equal(t, 0, client2.GetDeltaLen(), "no one-phase XA row events may be buffered") @@ -784,7 +783,7 @@ func TestGTIDClientXATransaction(t *testing.T) { // terminal XA COMMIT / XA ROLLBACK QueryEvent stays outside, under its // own GTID. The inner "XA START" QueryEvent must fail the payload before // the row events after it are buffered, surfacing exactly like the -// uncompressed abort: a checkpoint-preserving stream error. +// uncompressed abort: an unsupported-XA reason. // // Unique xids and per-run table names for the same reasons documented on // TestGTIDClientXATransaction. @@ -858,8 +857,8 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { // The prepare flushes the whole group as one compressed payload; the // guard must fail it at the inner "XA START" QueryEvent, ahead of the // inner row events. - require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, - 5*time.Second, 5*time.Millisecond, "a compressed XA group must fail the stream as a stream error") + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonUnsupportedXA) }, + 5*time.Second, 5*time.Millisecond, "a compressed XA group must fail the stream as unsupported XA") client.streamWG.Wait() // reader fully exited: buffering is final require.Equal(t, 0, client.GetDeltaLen(), "no row events inside the compressed XA group may be buffered") @@ -873,10 +872,9 @@ func TestGTIDClientXATransactionCompression(t *testing.T) { // TestGTIDClientXAGuardStream deterministically exercises the XA guard's // readStream wiring: any XA event must fail the stream via -// CancelFunc(FatalReasonStreamError) — before the XA transaction's row +// CancelFunc(FatalReasonUnsupportedXA) — before the XA transaction's row // events are buffered, and without promoting its GTID into the resume -// set (a resume must replay, and re-refuse, the XA group rather than -// skip it). +// set (the refused group is never treated as applied). // // Events are injected through a synthetic go-mysql BinlogStreamer rather // than a real server because the server writes an XA transaction's @@ -973,19 +971,19 @@ func TestGTIDClientXAGuardStream(t *testing.T) { require.NoError(t, streamer.AddEventToStreamer(ev)) } } - // expectAbort waits for the guard to fire as a stream error and for + // expectAbort waits for the guard to fire as unsupported XA and for // readStream to fully exit, then verifies nothing was buffered: no // row events, and no XA GTID in the resume set. expectAbort := func(t *testing.T, client *gtidClient, gotReason *atomic.Int64, gno int64) { t.Helper() - require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonStreamError) }, - 5*time.Second, 5*time.Millisecond, "the XA guard must fail the stream as a stream error") + require.Eventually(t, func() bool { return gotReason.Load() == int64(FatalReasonUnsupportedXA) }, + 5*time.Second, 5*time.Millisecond, "the XA guard must fail the stream as unsupported XA") client.streamWG.Wait() // reader fully exited: buffering is final require.Equal(t, 0, client.GetDeltaLen(), "no row events may be buffered once the guard fires") target, err := mysql.ParseMysqlGTIDSet(fmt.Sprintf("%s:%d", sid, gno)) require.NoError(t, err) require.False(t, client.getBufferedGTID().Contain(target), - "the XA GTID must not enter the resume set: a resume must replay (and re-refuse) the XA group") + "the XA GTID must not enter the resume set") } // The XA transaction's first group, exactly as the server writes it @@ -1067,6 +1065,30 @@ func TestGTIDProcessQueryEventXAGuard(t *testing.T) { } } +// A CTAS QueryEvent opens a transaction group: its GTID must stay pending, +// while its CREATE TABLE must still cancel a whole-schema subscriber. +func TestGTIDClientCTASNotifiesSchemaFilter(t *testing.T) { + empty, err := mysql.ParseMysqlGTIDSet("") + require.NoError(t, err) + sid := uuid.MustParse("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee") + var got []FatalReason + c := >idClient{ + logger: slog.Default(), + subs: newSubscriptionRegistry(), + bufferedGTID: empty, + flushedGTID: empty.Clone(), + ddlFilterSchema: "test", + callerCancelFunc: func(r FatalReason) bool { got = append(got, r); return true }, + pendingSID: sid[:], + pendingGNO: 7, + } + require.NoError(t, c.processQueryEvent(&replication.QueryEvent{Schema: []byte("test"), + Query: []byte("CREATE TABLE `ctas1` (`a` int NOT NULL) START TRANSACTION")})) + require.Equal(t, []FatalReason{FatalReasonSchemaChange}, got) + require.Equal(t, int64(7), c.pendingGNO, "CTAS must leave its GTID pending until XIDEvent") + require.Empty(t, c.getBufferedGTID().String(), "CTAS must not promote its GTID before row events") +} + // TestGTIDProcessTransactionPayloadXAGuard unit-tests the compressed // path directly: an inner "XA START" QueryEvent must fail the payload // before the row events after it are buffered, and an inner diff --git a/pkg/change/utils.go b/pkg/change/utils.go index 0d47404ca..0ff185a34 100644 --- a/pkg/change/utils.go +++ b/pkg/change/utils.go @@ -5,6 +5,7 @@ import ( "database/sql" "errors" "fmt" + "sync" "github.com/block/spirit/pkg/parser" "github.com/block/spirit/pkg/parser/ast" @@ -51,10 +52,18 @@ type queryEventInfo struct { xa bool } +// QueryEvents include BEGIN for nearly every transaction. Reuse parser +// instances to avoid allocating the parser's grammar cache for each event. +var queryEventParsers = sync.Pool{New: func() any { return parser.New() }} + // parseQueryEvent classifies transaction control and extracts DDL table names // from the same parse, so every consumer sees the same statement semantics. func parseQueryEvent(defaultSchema, statements string) (info queryEventInfo, err error) { - p := parser.New() + p := queryEventParsers.Get().(*parser.Parser) + defer func() { + p.Reset() + queryEventParsers.Put(p) + }() stmts, _, err := p.Parse(statements, "", "") if err != nil { return queryEventInfo{}, err @@ -235,6 +244,13 @@ func checkImmutableColumn(tbl *table.TableInfo, ordinal int, beforeRow, afterRow // any of the XA transaction's row events are buffered. var errXAUnsupported = errors.New("XA transactions detected in the binlog stream: spirit does not support XA workloads") +func fatalReasonForStreamError(err error) FatalReason { + if errors.Is(err, errXAUnsupported) { + return FatalReasonUnsupportedXA + } + return FatalReasonStreamError +} + // isMinimalRowImage returns true if the RowsEvent contains a minimal row image, // i.e. some columns were skipped. This happens when binlog_row_image=MINIMAL or NOBLOB. // With full row images, SkippedColumns entries are empty slices. diff --git a/pkg/datasync/runner.go b/pkg/datasync/runner.go index c08429950..5d565c6bf 100644 --- a/pkg/datasync/runner.go +++ b/pkg/datasync/runner.go @@ -1514,7 +1514,8 @@ func (r *Runner) startBackgroundRoutines(ctx context.Context) { // CancelFunc contract (change.ClientConfig), it fires for DDL detected on a // synced table (change.FatalReasonSchemaChange) AND for fatal stream errors // such as minimal RBR detection or exhausted streamer recreation attempts -// (change.FatalReasonStreamError). The reason names the trigger class in the +// (change.FatalReasonStreamError), or unsupported XA +// (change.FatalReasonUnsupportedXA). The reason names the trigger class in the // recorded error; the change client's logs carry the details. Unlike // migration/move, no resume state is invalidated for either reason: the // datasync checkpoint is Persistent and the caller decides fresh-vs-resume. diff --git a/pkg/datasync/sync_test.go b/pkg/datasync/sync_test.go index 0992a3bbd..13cb1d152 100644 --- a/pkg/datasync/sync_test.go +++ b/pkg/datasync/sync_test.go @@ -445,6 +445,13 @@ func TestFatalErrorConcurrentWithRunSetup(t *testing.T) { require.LessOrEqual(t, cancelCalls.Load(), int64(1)) } +func TestFatalErrorUnsupportedXAReason(t *testing.T) { + runner, err := NewRunner(&Sync{}) + require.NoError(t, err) + require.True(t, runner.fatalError(change.FatalReasonUnsupportedXA)) + require.ErrorContains(t, runner.fatal(), "unsupported-xa") +} + // TestSyncResume verifies that the initial copy writes a copier-watermark // checkpoint and that a second run against the same (non-empty) target detects // it and resumes — opening the chunker at the saved watermark instead of diff --git a/pkg/migration/runner.go b/pkg/migration/runner.go index 5c4d654a4..039d4b9bd 100644 --- a/pkg/migration/runner.go +++ b/pkg/migration/runner.go @@ -1375,6 +1375,9 @@ func (r *Runner) fatalError(reason change.FatalReason) bool { // changed, so the checkpoint remains valid. Keep it and tell the // operator how to recover. r.logger.Error("fatal replication stream error; the checkpoint has been preserved — re-run spirit to resume the migration from it") + case change.FatalReasonUnsupportedXA: + r.logger.Error("XA transaction detected; the checkpoint will be invalidated — stop XA activity and start a fresh migration") + fallthrough default: // Schema change — and, defensively, any future reason we don't // recognize (invalidating is the safe default: it costs a restart, diff --git a/pkg/migration/runner_close_test.go b/pkg/migration/runner_close_test.go index d670cf8f1..15f9b346b 100644 --- a/pkg/migration/runner_close_test.go +++ b/pkg/migration/runner_close_test.go @@ -133,4 +133,17 @@ func TestFatalErrorReasonCheckpointHandling(t *testing.T) { require.True(t, checkpointTableExists(t, r), "a stream-error fatal must preserve the checkpoint table so the migration can resume") }) + + t.Run("UnsupportedXADropsCheckpoint", func(t *testing.T) { + t.Parallel() + r := setupRunnerForChecksumTest(t, "fatal_reason_xa") + var cancelCalls atomic.Int32 + r.cancelFunc = func() { cancelCalls.Add(1) } + + require.True(t, r.fatalError(change.FatalReasonUnsupportedXA)) + require.Equal(t, status.ErrCleanup, r.status.Get()) + require.Equal(t, int32(1), cancelCalls.Load()) + require.False(t, checkpointTableExists(t, r), + "a checkpoint that replays the refused XA group cannot be resumed") + }) } diff --git a/pkg/move/runner.go b/pkg/move/runner.go index f43933121..9abea86f2 100644 --- a/pkg/move/runner.go +++ b/pkg/move/runner.go @@ -1534,6 +1534,9 @@ func (r *Runner) fatalError(reason change.FatalReason) bool { // changed, so the checkpoint remains valid. Keep it and tell the // operator how to recover. r.logger.Error("fatal replication stream error; the checkpoint has been preserved — re-run spirit to resume the move from it") + case change.FatalReasonUnsupportedXA: + r.logger.Error("XA transaction detected; the checkpoint will be invalidated — stop XA activity and start a fresh move") + fallthrough default: // Schema change — and, defensively, any future reason we don't // recognize (invalidating is the safe default: it costs a restart, diff --git a/pkg/move/runner_close_test.go b/pkg/move/runner_close_test.go index 859e7e1ea..a30b07a4f 100644 --- a/pkg/move/runner_close_test.go +++ b/pkg/move/runner_close_test.go @@ -155,6 +155,15 @@ func TestFatalErrorReasonCheckpointHandling(t *testing.T) { require.True(t, checkpointTableExists(t, r), "a stream-error fatal must preserve the checkpoint table so the move can resume") }) + + t.Run("UnsupportedXADropsCheckpoint", func(t *testing.T) { + r, cancelCalls := makeRunner(t) + require.True(t, r.fatalError(change.FatalReasonUnsupportedXA)) + require.Equal(t, status.ErrCleanup, r.status.Get()) + require.Equal(t, int32(1), cancelCalls.Load()) + require.False(t, checkpointTableExists(t, r), + "a checkpoint that replays the refused XA group cannot be resumed") + }) } // fakeChangeSource is a minimal change.Source used to observe that