From bcf0c1e7dd79063f8ebe01925d0a92f24e584acd Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 2 Jul 2026 06:44:01 -0600 Subject: [PATCH 1/2] fix(change): survive uint32 LogPos wraparound in the replay-skip guard MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ev.Header.LogPos is a 4-byte wire field, and a binlog file only rotates on a transaction boundary: one transaction writing more than 4GiB of row images past the max_binlog_size point pushes end positions past 2^32, where they wrap to small values. setBufferedPos's monotonic guard then freezes bufferedPos just under the wrap point, and every subsequent RowsEvent in the file compared <= bufferedPos — silently discarded as a post-reconnect replay. Lost changes were only caught by the checksum (where enabled). Make the skip decision wraparound-aware and fail safe (re-apply instead of skip): outside a post-recreateStreamer replay window, a same-file event more than 2^31 below bufferedPos is classified as wraparound, logged once per file, and delivered. Re-delivery is idempotent (REPLACE/DELETE); skipping is not. The replay window is a local flag in readStream, set on a successful streamer recreation and cleared when the first rows event is delivered again, so normal reconnect-replay deduplication is unchanged. Co-Authored-By: Claude Fable 5 --- pkg/change/binlog.go | 87 +++++++++++++++- pkg/change/binlog_test.go | 205 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 290 insertions(+), 2 deletions(-) diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go index 3978bb57..5f9e2e0b 100644 --- a/pkg/change/binlog.go +++ b/pkg/change/binlog.go @@ -190,6 +190,53 @@ func shouldSkipReplayedEvent(eventPos, bufferedPos mysql.Position) bool { return eventPos.Compare(bufferedPos) <= 0 } +// logPosWrapThreshold is how far below bufferedPos (within the same binlog +// file) an event's end position must sit before the gap is classified as +// uint32 LogPos wraparound rather than a post-reconnect replay. Half the +// uint32 space cleanly separates the two: immediately after a wrap the gap +// is nearly 2^32 (bufferedPos froze just under 4GiB, wrapped positions +// restart near zero), while a replay's gap only ever spans event positions +// that were actually reached within the file. +const logPosWrapThreshold = uint32(1) << 31 // 2 GiB + +// isWrappedLogPos reports whether eventPos sitting below bufferedPos in the +// same binlog file is best explained by LogPos wraparound. ev.Header.LogPos +// is a 4-byte wire field, and a binlog file only rotates on a transaction +// boundary — one transaction writing more than 4GiB of row images past the +// max_binlog_size point pushes end positions past 2^32, where they wrap to +// small values. setBufferedPos's monotonicity then freezes bufferedPos just +// under the wrap point, so every post-wrap event compares below it by an +// enormous margin. +func isWrappedLogPos(eventPos, bufferedPos mysql.Position) bool { + return eventPos.Name == bufferedPos.Name && + eventPos.Pos < bufferedPos.Pos && + bufferedPos.Pos-eventPos.Pos > logPosWrapThreshold +} + +// shouldSkipRowsEvent is the full skip decision for a RowsEvent: skip +// genuine replays (shouldSkipReplayedEvent), but fail safe on LogPos +// wraparound by delivering the event. wrapped=true reports that the event +// compares below bufferedPos only because the 4-byte LogPos wrapped past +// 4GiB — the event is live, and skipping it would silently discard row +// changes. Re-delivering is safe (REPLACE/DELETE application is idempotent); +// skipping is not. +// +// replaying must be true from a successful recreateStreamer until the replay +// catches back up to bufferedPos: replayed events legitimately sit far below +// the high-water mark (the replay restarts at position 4), so the wraparound +// heuristic is suppressed inside that window. The residual ambiguity — a +// reconnect while bufferedPos is already frozen at a pre-wrap value — cannot +// be resolved with 32-bit positions at all, and keeps the old skip behavior. +func shouldSkipRowsEvent(eventPos, bufferedPos mysql.Position, replaying bool) (skip, wrapped bool) { + if !shouldSkipReplayedEvent(eventPos, bufferedPos) { + return false, false + } + if !replaying && isWrappedLogPos(eventPos, bufferedPos) { + return false, true + } + return true, false +} + // AllChangesFlushed returns true if all buffered changes across all // subscriptions have been flushed to the target tables. // Satisfies Source interface. @@ -452,6 +499,17 @@ func (c *binlogClient) readStream(ctx context.Context) { lastErrorTime := time.Time{} var recentErrors []string // Track recent errors for debugging + // replaying is true from a successful streamer recreation until the + // replay catches back up to the buffered position (i.e. the first rows + // event delivered again). Inside the window, events at or below + // bufferedPos are replays and must be skipped; outside it, a same-file + // event far below bufferedPos is uint32 LogPos wraparound and must be + // delivered instead. See shouldSkipRowsEvent. + replaying := false + // wrapWarnedFile dedupes the LogPos-wraparound warning to once per + // binlog file. + wrapWarnedFile := "" + c.logger.Debug("readStream started for binlog position", "position", startPos, "log_name", currentLogName) for { @@ -546,6 +604,11 @@ func (c *binlogClient) readStream(ctx context.Context) { consecutiveErrors = 0 recreateAttempts = 0 backoffDuration = initialBackoffDuration + // The recreated dump restarts at position 4 of the + // current file and replays events we already buffered. + // Mark the replay window so those low positions are + // skipped as replays, not mistaken for LogPos wraparound. + replaying = true } lastErrorTime = currentTime } @@ -587,13 +650,33 @@ func (c *binlogClient) readStream(ctx context.Context) { case *replication.RowsEvent: // Skip events already buffered/applied: a post-recreateStreamer // replay must not re-buffer them and regress a key to a stale - // image. See shouldSkipReplayedEvent. + // image. LogPos is only 4 bytes on the wire, though, and one + // huge transaction can push a file past 4GiB (rotation only + // happens on a transaction boundary), wrapping positions back + // to small values — those events are live, not replays, and + // must be delivered. See shouldSkipRowsEvent. eventPos := mysql.Position{Name: currentLogName, Pos: ev.Header.LogPos} - if shouldSkipReplayedEvent(eventPos, c.getBufferedPos()) { + bufferedPos := c.getBufferedPos() + skip, wrapped := shouldSkipRowsEvent(eventPos, bufferedPos, replaying) + if wrapped && wrapWarnedFile != currentLogName { + wrapWarnedFile = currentLogName + c.logger.Warn("binlog LogPos wrapped past 4GiB within a single file; "+ + "delivering events below the buffered position instead of skipping them as replays. "+ + "The buffered/flushed position cannot advance until the next rotation.", + "file", currentLogName, + "event_log_pos", ev.Header.LogPos, + "buffered_position", bufferedPos) + } + if skip { c.logger.Debug("skipping replayed rows event at or below the buffered position", "event_position", eventPos) continue } + // This event is being delivered, so any post-recreate replay + // window is over: the replay has caught back up to bufferedPos + // (or the position wrapped, which the window cannot outlive — + // reaching a wrap requires ~4GiB of delivered events first). + replaying = false if err = c.processRowsEvent(ev, event); err != nil { c.logger.Error("fatal error processing binlog rows event", "error", err) c.fatalError() diff --git a/pkg/change/binlog_test.go b/pkg/change/binlog_test.go index ca31c5c2..83373dfd 100644 --- a/pkg/change/binlog_test.go +++ b/pkg/change/binlog_test.go @@ -1,10 +1,12 @@ package change import ( + "bytes" "context" "database/sql" "fmt" "log/slog" + "strings" "sync" "testing" "time" @@ -16,6 +18,7 @@ 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/stretchr/testify/require" "go.uber.org/goleak" @@ -967,6 +970,208 @@ func TestShouldSkipReplayedEventWidensWithBufferedPos(t *testing.T) { "once the boundary advances, later live events are delivered") } +// TestShouldSkipRowsEventWraparound unit-tests the wraparound-aware skip +// decision. ev.Header.LogPos is a 4-byte wire field: one transaction writing +// more than 4GiB of row images into a single binlog file (rotation only +// happens on a transaction boundary) wraps end positions back to small +// values. setBufferedPos's monotonicity freezes bufferedPos just under the +// wrap point, so pre-fix every post-wrap RowsEvent compared <= bufferedPos +// and was silently discarded as a replay. The fix fails safe: outside a +// post-reconnect replay window, a same-file event below bufferedPos by more +// than 2^31 is classified as wraparound and delivered (re-applying is +// idempotent; skipping loses changes). +func TestShouldSkipRowsEventWraparound(t *testing.T) { + const file = "binlog.000010" + // bufferedPos as left behind by a wrap: frozen just under 4GiB. + frozen := mysql.Position{Name: file, Pos: 4294967000} + tests := []struct { + name string + eventPos mysql.Position + bufferedPos mysql.Position + replaying bool + skip bool + wrapped bool + }{ + { + name: "live event above bufferedPos: deliver", + eventPos: mysql.Position{Name: file, Pos: 4294967100}, + bufferedPos: frozen, + skip: false, + }, + { + name: "live event slightly below bufferedPos: replay dedup skips (unchanged behavior)", + eventPos: mysql.Position{Name: file, Pos: frozen.Pos - 1000}, + bufferedPos: frozen, + skip: true, + }, + { + name: "live event equal to bufferedPos: skip (last buffered event)", + eventPos: frozen, + bufferedPos: frozen, + skip: true, + }, + { + name: "gap of exactly 2^31 is not classified as wraparound: skip", + eventPos: mysql.Position{Name: file, Pos: frozen.Pos - logPosWrapThreshold}, + bufferedPos: frozen, + skip: true, + }, + { + name: "gap just past 2^31: wraparound, deliver", + eventPos: mysql.Position{Name: file, Pos: frozen.Pos - logPosWrapThreshold - 1}, + bufferedPos: frozen, + skip: false, + wrapped: true, + }, + { + name: "realistic post-wrap event near the start of the uint32 space: wraparound, deliver", + eventPos: mysql.Position{Name: file, Pos: 500}, + bufferedPos: frozen, + skip: false, + wrapped: true, + }, + { + name: "earlier file with a huge offset gap is not wraparound: skip", + eventPos: mysql.Position{Name: "binlog.000009", Pos: 500}, + bufferedPos: frozen, + skip: true, + }, + { + name: "replay window: event below bufferedPos is a replay, skip", + eventPos: mysql.Position{Name: file, Pos: frozen.Pos - 1000}, + bufferedPos: frozen, + replaying: true, + skip: true, + }, + { + name: "replay window suppresses the wraparound heuristic: a replayed " + + "event far below the high-water mark (replay restarts at pos 4) " + + "must still be skipped", + eventPos: mysql.Position{Name: file, Pos: 500}, + bufferedPos: frozen, + replaying: true, + skip: true, + }, + { + name: "replay catch-up: event above bufferedPos delivers even while replaying", + eventPos: mysql.Position{Name: file, Pos: 4294967100}, + bufferedPos: frozen, + replaying: true, + skip: false, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + skip, wrapped := shouldSkipRowsEvent(tt.eventPos, tt.bufferedPos, tt.replaying) + require.Equal(t, tt.skip, skip, "skip") + require.Equal(t, tt.wrapped, wrapped, "wrapped") + }) + } +} + +// TestReadStreamDeliversWrappedLogPosEvents drives readStream through an +// injected streamer (no real syncer) to lock in the end-to-end wraparound +// behavior: with bufferedPos frozen just under 4GiB by a >4GiB transaction, +// post-wrap RowsEvents (small LogPos) must be buffered — pre-fix they were +// silently discarded as replays — and the wraparound warning is logged once +// per file. An event only slightly below bufferedPos (a genuine stale +// position) must still be skipped. +func TestReadStreamDeliversWrappedLogPosEvents(t *testing.T) { + t1 := `CREATE TABLE subscription_test ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + t2 := `CREATE TABLE _subscription_test_new ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + srcTable, dstTable := setupTestTables(t, t1, t2) + + var logBuf bytes.Buffer + logger := slog.New(slog.NewTextHandler(&logBuf, &slog.HandlerOptions{Level: slog.LevelDebug})) + + // bufferedPos frozen just under the 4GiB wrap point, as setBufferedPos's + // monotonic guard leaves it once LogPos wraps. + frozen := mysql.Position{Name: "binlog.000042", Pos: 4294967000} + streamer := replication.NewBinlogStreamer() + client := &binlogClient{ + logger: logger, + subs: newSubscriptionRegistry(), + streamer: streamer, + flushedPos: frozen, + bufferedPos: frozen, + } + sub := &bufferedMap{ + logger: logger, + table: srcTable, + newTable: dstTable, + changes: make(map[string]bufferedChange), + pkIsMemoryComparable: true, + } + sub.cond = sync.NewCond(&sub.Mutex) + require.True(t, client.subs.Add(encodeSchemaTable(srcTable.SchemaName, srcTable.TableName), sub)) + + mkInsert := func(logPos uint32, pk int) *replication.BinlogEvent { + return &replication.BinlogEvent{ + Header: &replication.EventHeader{ + Timestamp: uint32(time.Now().Unix()), //nolint: gosec + EventType: replication.WRITE_ROWS_EVENTv2, + LogPos: logPos, + }, + Event: &replication.RowsEvent{ + Table: &replication.TableMapEvent{ + Schema: []byte(srcTable.SchemaName), + Table: []byte(srcTable.TableName), + }, + Rows: [][]any{{pk, "x"}}, + }, + } + } + // A genuinely stale position (slightly below the high-water mark, gap + // far under 2^31): must be skipped as a replay. + require.NoError(t, streamer.AddEventToStreamer(mkInsert(frozen.Pos-1000, 1))) + // Two post-wrap events (LogPos wrapped back to small values): must be + // delivered, with the warning logged only once for the file. + require.NoError(t, streamer.AddEventToStreamer(mkInsert(500, 2))) + require.NoError(t, streamer.AddEventToStreamer(mkInsert(600, 3))) + + ctx, cancel := context.WithCancel(t.Context()) + client.streamWG.Add(1) + go client.readStream(ctx) + + // Events are processed in order, so once the last event's key is + // buffered, the decisions for all three are final. + key3 := utils.HashKey([]any{3}) + require.Eventually(t, func() bool { + sub.Lock() + defer sub.Unlock() + _, ok := sub.changes[key3] + return ok + }, 10*time.Second, 10*time.Millisecond, "post-wrap events were not delivered") + + cancel() + client.streamWG.Wait() + + sub.Lock() + _, hasKey1 := sub.changes[utils.HashKey([]any{1})] + _, hasKey2 := sub.changes[utils.HashKey([]any{2})] + sub.Unlock() + require.False(t, hasKey1, "an event slightly below bufferedPos is a genuine replay and must still be skipped") + require.True(t, hasKey2, "post-wrap events must be buffered, not discarded as replays") + require.Equal(t, 2, sub.Length()) + + // bufferedPos stays frozen at the pre-wrap value (setBufferedPos is + // monotonic and wrapped positions compare below it). + require.Equal(t, frozen, client.getBufferedPos()) + + // The wraparound warning is emitted exactly once per binlog file, even + // though two wrapped events were delivered. + require.Equal(t, 1, strings.Count(logBuf.String(), "wrapped past 4GiB"), + "expected exactly one wraparound warning per file") +} + // TestRecreateStreamerSkipsFlushedReplay locks in the fix end-to-end by // driving the real consecutive-error path that calls recreateStreamer: // 1. Stream an INSERT + UPDATE for one PK and flush, so the target holds From a74678412b3a7a71a5c7518f1756493312eb9e1b Mon Sep 17 00:00:00 2001 From: Morgan Tocker Date: Thu, 2 Jul 2026 06:47:31 -0600 Subject: [PATCH 2/2] fix(change): hard-fail on unknown rows-event subtypes instead of dropping rows MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit parseEventType does not recognize e.g. PARTIAL_UPDATE_ROWS_EVENT, which go-mysql delivers as a regular *replication.RowsEvent. It reached processRowsEvent (both the binlog and GTID clients), fell into the default branch, and its rows were skipped with only logger.Error("unknown event type") — silent data loss, asymmetric with the minimal-row-image case which hard-fails. Preflight requires binlog_row_value_options to be empty, but the global can change after preflight, at which point JSON partial updates stream and are silently not applied. Return an error naming the event type (string form and header value) and the table from processRowsEvent in both clients, so it flows through the existing fatalError path exactly like the minimal-row-image case. The insert/delete loop's default branch becomes a hard error too, so a future eventType addition cannot reintroduce a silent drop. Co-Authored-By: Claude Fable 5 --- pkg/change/binlog.go | 18 ++++++++-- pkg/change/binlog_test.go | 70 +++++++++++++++++++++++++++++++++++++++ pkg/change/gtid.go | 14 ++++++-- pkg/change/gtid_test.go | 68 +++++++++++++++++++++++++++++++++++++ 4 files changed, 166 insertions(+), 4 deletions(-) diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go index 5f9e2e0b..6ab814bc 100644 --- a/pkg/change/binlog.go +++ b/pkg/change/binlog.go @@ -796,8 +796,19 @@ func (c *binlogClient) processRowsEvent(ev *replication.BinlogEvent, e *replicat return fmt.Errorf("received a minimal RBR event for table %s.%s, but we require binlog_row_image=FULL on the source server", string(e.Table.Schema), string(e.Table.Table)) } - tbl := sub.Tables()[0] eventType := parseEventType(ev.Header.EventType) + if eventType == eventTypeUnknown { + // Hard-fail, mirroring the minimal-row-image check above: an + // unrecognized rows-event subtype (e.g. PARTIAL_UPDATE_ROWS_EVENT, + // which the server emits once binlog_row_value_options=PARTIAL_JSON + // is set — the global can change after our preflight check) still + // carries row changes. Dropping it with only an error log would + // silently lose those changes; they'd only be caught by the + // checksum where enabled. + return fmt.Errorf("received unsupported rows event type %v (0x%02x) for table %s.%s: cannot apply its row changes", ev.Header.EventType, uint8(ev.Header.EventType), string(e.Table.Schema), string(e.Table.Table)) + } + + tbl := sub.Tables()[0] // Decode ENUM ordinals / SET bitmasks back to their string form and // re-pad BINARY(N) values (MySQL strips trailing 0x00 from the row @@ -852,7 +863,10 @@ func (c *binlogClient) processRowsEvent(ev *replication.BinlogEvent, e *replicat case eventTypeDelete: sub.HasChanged(key, nil, true) default: - c.logger.Error("unknown event type", "type", ev.Header.EventType) + // Unreachable: eventTypeUnknown is rejected above and + // eventTypeUpdate returned earlier. Kept as a hard error so a + // future eventType addition cannot silently drop rows. + return fmt.Errorf("unhandled rows event type %v (0x%02x) for table %s.%s", ev.Header.EventType, uint8(ev.Header.EventType), string(e.Table.Schema), string(e.Table.Table)) } } return nil diff --git a/pkg/change/binlog_test.go b/pkg/change/binlog_test.go index 83373dfd..f2d5c8a7 100644 --- a/pkg/change/binlog_test.go +++ b/pkg/change/binlog_test.go @@ -1172,6 +1172,76 @@ func TestReadStreamDeliversWrappedLogPosEvents(t *testing.T) { "expected exactly one wraparound warning per file") } +// TestProcessRowsEventUnknownSubtypeFails asserts that a rows-event subtype +// parseEventType does not recognize is a hard error, not a logged skip. +// go-mysql delivers e.g. PARTIAL_UPDATE_ROWS_EVENT (emitted once +// binlog_row_value_options=PARTIAL_JSON is set — the global can change after +// preflight) as a regular *replication.RowsEvent; pre-fix processRowsEvent +// dropped its rows with only logger.Error, silently losing changes. It must +// fail exactly like the minimal-row-image case so readStream cancels the +// migration via fatalError. +func TestProcessRowsEventUnknownSubtypeFails(t *testing.T) { + t1 := `CREATE TABLE subscription_test ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + t2 := `CREATE TABLE _subscription_test_new ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + srcTable, dstTable := setupTestTables(t, t1, t2) + + client := &binlogClient{ + logger: slog.Default(), + subs: newSubscriptionRegistry(), + } + sub := &bufferedMap{ + logger: client.logger, + table: srcTable, + newTable: dstTable, + changes: make(map[string]bufferedChange), + pkIsMemoryComparable: true, + } + sub.cond = sync.NewCond(&sub.Mutex) + require.True(t, client.subs.Add(encodeSchemaTable(srcTable.SchemaName, srcTable.TableName), sub)) + + mkEvent := func(et replication.EventType, schema, tableName string) (*replication.BinlogEvent, *replication.RowsEvent) { + rows := &replication.RowsEvent{ + Table: &replication.TableMapEvent{ + Schema: []byte(schema), + Table: []byte(tableName), + }, + Rows: [][]any{{1, "x"}}, + } + return &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: et, LogPos: 1000}, + Event: rows, + }, rows + } + + // Sanity: a recognized subtype flows through and buffers the row. + ev, rows := mkEvent(replication.WRITE_ROWS_EVENTv2, srcTable.SchemaName, srcTable.TableName) + require.NoError(t, client.processRowsEvent(ev, rows)) + require.Equal(t, 1, sub.Length()) + + // An unrecognized subtype for a subscribed table must hard-error, + // naming the event type (string and header value) and the table. + ev, rows = mkEvent(replication.PARTIAL_UPDATE_ROWS_EVENT, srcTable.SchemaName, srcTable.TableName) + err := client.processRowsEvent(ev, rows) + require.Error(t, err) + require.ErrorContains(t, err, "PartialUpdateRowsEvent") + require.ErrorContains(t, err, fmt.Sprintf("0x%02x", uint8(replication.PARTIAL_UPDATE_ROWS_EVENT))) + require.ErrorContains(t, err, srcTable.TableName) + require.Equal(t, 1, sub.Length(), "the unsupported event must not buffer rows") + + // Events for tables without a subscription stay ignored regardless of + // subtype (e.g. writes to unrelated tables, or to a _new table). + ev, rows = mkEvent(replication.PARTIAL_UPDATE_ROWS_EVENT, srcTable.SchemaName, "not_subscribed") + require.NoError(t, client.processRowsEvent(ev, rows)) +} + // TestRecreateStreamerSkipsFlushedReplay locks in the fix end-to-end by // driving the real consecutive-error path that calls recreateStreamer: // 1. Stream an INSERT + UPDATE for one PK and flush, so the target holds diff --git a/pkg/change/gtid.go b/pkg/change/gtid.go index 542bd3fa..e632a77b 100644 --- a/pkg/change/gtid.go +++ b/pkg/change/gtid.go @@ -629,8 +629,15 @@ func (c *gtidClient) processRowsEvent(ev *replication.BinlogEvent, e *replicatio return fmt.Errorf("received a minimal RBR event for table %s.%s, but we require binlog_row_image=FULL on the source server", string(e.Table.Schema), string(e.Table.Table)) } - tbl := sub.Tables()[0] eventType := parseEventType(ev.Header.EventType) + if eventType == eventTypeUnknown { + // Hard-fail, mirroring the minimal-row-image check above — see the + // matching guard in binlogClient.processRowsEvent. Dropping an + // unrecognized rows-event subtype would silently lose row changes. + return fmt.Errorf("received unsupported rows event type %v (0x%02x) for table %s.%s: cannot apply its row changes", ev.Header.EventType, uint8(ev.Header.EventType), string(e.Table.Schema), string(e.Table.Table)) + } + + tbl := sub.Tables()[0] // Decode ENUM/SET integers and re-pad BINARY(N) values before key // extraction and buffering — see the matching block in binlog.go's @@ -676,7 +683,10 @@ func (c *gtidClient) processRowsEvent(ev *replication.BinlogEvent, e *replicatio case eventTypeDelete: sub.HasChanged(key, nil, true) default: - c.logger.Error("unknown event type", "type", ev.Header.EventType) + // Unreachable: eventTypeUnknown is rejected above and + // eventTypeUpdate returned earlier. Kept as a hard error so a + // future eventType addition cannot silently drop rows. + return fmt.Errorf("unhandled rows event type %v (0x%02x) for table %s.%s", ev.Header.EventType, uint8(ev.Header.EventType), string(e.Table.Schema), string(e.Table.Table)) } } return nil diff --git a/pkg/change/gtid_test.go b/pkg/change/gtid_test.go index 773f175c..8ea3cbb4 100644 --- a/pkg/change/gtid_test.go +++ b/pkg/change/gtid_test.go @@ -3,6 +3,7 @@ package change import ( "fmt" "log/slog" + "sync" "testing" "github.com/block/spirit/pkg/applier" @@ -11,6 +12,7 @@ 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" @@ -349,3 +351,69 @@ func TestGTIDRoundtripPosition(t *testing.T) { require.NoError(t, client2.StartFromPosition(t.Context(), pos)) client2.Close() } + +// TestGTIDProcessRowsEventUnknownSubtypeFails mirrors +// TestProcessRowsEventUnknownSubtypeFails for the GTID client: a rows-event +// subtype parseEventType does not recognize must be a hard error (routed to +// fatalError by readStream), never a logged skip that silently loses rows. +func TestGTIDProcessRowsEventUnknownSubtypeFails(t *testing.T) { + t1 := `CREATE TABLE subscription_test ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + t2 := `CREATE TABLE _subscription_test_new ( + id INT NOT NULL, + name VARCHAR(255) NOT NULL, + PRIMARY KEY (id) + )` + srcTable, dstTable := setupTestTables(t, t1, t2) + + client := >idClient{ + logger: slog.Default(), + subs: newSubscriptionRegistry(), + } + sub := &bufferedMap{ + logger: client.logger, + table: srcTable, + newTable: dstTable, + changes: make(map[string]bufferedChange), + pkIsMemoryComparable: true, + } + sub.cond = sync.NewCond(&sub.Mutex) + require.True(t, client.subs.Add(encodeSchemaTable(srcTable.SchemaName, srcTable.TableName), sub)) + + mkEvent := func(et replication.EventType, schema, tableName string) (*replication.BinlogEvent, *replication.RowsEvent) { + rows := &replication.RowsEvent{ + Table: &replication.TableMapEvent{ + Schema: []byte(schema), + Table: []byte(tableName), + }, + Rows: [][]any{{1, "x"}}, + } + return &replication.BinlogEvent{ + Header: &replication.EventHeader{EventType: et, LogPos: 1000}, + Event: rows, + }, rows + } + + // Sanity: a recognized subtype flows through and buffers the row. + ev, rows := mkEvent(replication.WRITE_ROWS_EVENTv2, srcTable.SchemaName, srcTable.TableName) + require.NoError(t, client.processRowsEvent(ev, rows)) + require.Equal(t, 1, sub.Length()) + + // An unrecognized subtype for a subscribed table must hard-error, + // naming the event type (string and header value) and the table. + ev, rows = mkEvent(replication.PARTIAL_UPDATE_ROWS_EVENT, srcTable.SchemaName, srcTable.TableName) + err := client.processRowsEvent(ev, rows) + require.Error(t, err) + require.ErrorContains(t, err, "PartialUpdateRowsEvent") + require.ErrorContains(t, err, fmt.Sprintf("0x%02x", uint8(replication.PARTIAL_UPDATE_ROWS_EVENT))) + require.ErrorContains(t, err, srcTable.TableName) + require.Equal(t, 1, sub.Length(), "the unsupported event must not buffer rows") + + // Events for tables without a subscription stay ignored regardless of + // subtype (e.g. writes to unrelated tables, or to a _new table). + ev, rows = mkEvent(replication.PARTIAL_UPDATE_ROWS_EVENT, srcTable.SchemaName, "not_subscribed") + require.NoError(t, client.processRowsEvent(ev, rows)) +}