diff --git a/pkg/change/binlog.go b/pkg/change/binlog.go
index 3978bb57..6ab814bc 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()
@@ -713,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
@@ -769,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 ca31c5c2..f2d5c8a7 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,278 @@ 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")
+}
+
+// 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 := &gtidClient{
+		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))
+}