fix(mysqlcommon): flush binlog event at marker

- Flush a completed row event at each "# at" marker instead of waiting
  for the next ### op-line or EOF. In an unbounded CDC stream the next
  op-line may never arrive during a quiet period, so the last change
  would otherwise stay buffered and never reach the target.
- The marker is the existing event boundary already used for the bounded
  stopAt check; flushing there is safe and does not double-emit (flush
  nils the pending event).
- Add a regression test asserting events flush at the following marker.

Author: nixrajput

internal/driver/_mysqlcommon/changestream.go

@@ -179,14 +179,18 @@ func parseBinlogRows(r io.Reader, meta *tableMetaCache, emit func(canonical.Cano
 		// Track the current binlog offset from "# at <pos>" markers.
 		if p, ok := parseAtMarker(line); ok {
 			pos.Position = p
-			// Bounded (incremental) stop: an "# at" marker precedes each event, so
-			// once the marker reaches/passes the captured end position in the end
-			// file, every earlier event has been parsed. Flush the pending event
-			// and return at this clean boundary.
+			// An "# at" marker precedes each event, so it is the boundary at which
+			// the preceding event is complete. Flush the pending event here so it
+			// is emitted promptly — without this, in an unbounded CDC stream the
+			// last change before a quiet period would sit buffered until the next
+			// event (or EOF, which never comes), never reaching the target.
+			if err := flush(); err != nil {
+				return pos, err
+			}
+			// Bounded (incremental) stop: once the marker reaches/passes the
+			// captured end position in the end file, every earlier event has been
+			// parsed and flushed above, so return at this clean boundary.
 			if stopAt != nil && pos.File == stopAt.File && pos.Position >= stopAt.Position {
-				if err := flush(); err != nil {
-					return pos, err
-				}
 				return pos, nil
 			}
 			continue

internal/driver/_mysqlcommon/changestream_test.go

@@ -155,3 +155,46 @@ func TestParseBinlogRows_NullVsNullString(t *testing.T) {
 		t.Errorf("quoted 'NULL': name = %#v, want \"NULL\" string", got[1].Values["name"])
 	}
 }
+
+// TestParseBinlogRows_FlushesAtMarker proves a completed event is emitted at the
+// next "# at" marker rather than being held until a following ### op-line. In an
+// unbounded CDC stream the next op-line may never arrive during a quiet period,
+// so without the marker-boundary flush the last change would sit buffered and
+// never reach the target. The emit callback fails if it ever sees a pending
+// event still buffered when the NEXT marker is processed — i.e. it asserts each
+// event is flushed by the marker that immediately follows it, not deferred.
+func TestParseBinlogRows_FlushesAtMarker(t *testing.T) {
+	// Two events separated by markers; the second is followed by a trailing
+	// marker. If flushing only happened on the next op-line, the second event
+	// would emit at end-of-loop, after the "# at 300" marker was already seen.
+	transcript := `# at 4
+### INSERT INTO ` + "`test`.`widgets`" + `
+### SET
+###   @1=7
+###   @2='a'
+# at 100
+### INSERT INTO ` + "`test`.`widgets`" + `
+### SET
+###   @1=8
+###   @2='b'
+# at 300
+`
+	meta := &tableMetaCache{cache: map[string]*tableLayout{
+		"widgets": {cols: []string{"id", "name"}, pk: map[string]bool{"id": true}},
+	}}
+
+	var got []canonical.CanonicalChange
+	emit := func(ch canonical.CanonicalChange) error { got = append(got, ch); return nil }
+
+	if _, err := parseBinlogRows(strings.NewReader(transcript), meta, emit, BinlogPosition{File: "mysql-bin.000001"}, nil); err != nil {
+		t.Fatalf("parseBinlogRows: %v", err)
+	}
+	if len(got) != 2 {
+		t.Fatalf("got %d changes, want 2: %+v", len(got), got)
+	}
+	// The first event must be emitted before the second begins — proving the
+	// "# at 100" marker flushed it rather than the second INSERT's op-line.
+	if got[0].Values["name"] != "a" || got[1].Values["name"] != "b" {
+		t.Errorf("changes out of order or wrong: %+v", got)
+	}
+}