GTID 1/5: Introduce BinlogCoordinate abstraction (file/position backed)

binlog_coordinate.go

@@ -0,0 +1,178 @@
+package ghostferry
+
+import (
+	"encoding/json"
+	"fmt"
+
+	"github.com/go-mysql-org/go-mysql/mysql"
+)
+
+// BinlogCoordinateType identifies how a binlog position is expressed.
+//
+// This is the seam that lets Ghostferry evolve from file/position based
+// replication coordinates towards GTID based coordinates without forcing every
+// consumer to know which representation is in use. In this first iteration only
+// the file/position representation is implemented; the GTID representation is
+// reserved so that state serialized today remains forward compatible.
+type BinlogCoordinateType string
+
+const (
+	// BinlogCoordinateFilePosition is the classic (file, position) coordinate.
+	BinlogCoordinateFilePosition BinlogCoordinateType = "file_position"
+
+	// BinlogCoordinateGTID is reserved for the future GTID based coordinate.
+	// It is intentionally defined now so that the serialization format and the
+	// strategy dispatch have a stable name to target.
+	BinlogCoordinateGTID BinlogCoordinateType = "gtid"
+)
+
+// BinlogCoordinate is a representation-agnostic replication coordinate.
+//
+// Today it only wraps a file/position (mysql.Position). It exists so that the
+// rest of Ghostferry can be migrated to talk in terms of "a coordinate" rather
+// than "a file and a position", which is a prerequisite for adding a GTID mode
+// behind a feature flag without a second invasive refactor.
+//
+// The zero value is a zero file/position coordinate, matching the previous
+// behavior where an empty mysql.Position was used as the "no coordinate"
+// sentinel.
+type BinlogCoordinate struct {
+	// Type selects the active representation. An empty Type is treated as
+	// BinlogCoordinateFilePosition for backwards compatibility.
+	Type BinlogCoordinateType
+
+	// FilePosition holds the coordinate when Type is BinlogCoordinateFilePosition.
+	FilePosition mysql.Position
+}
+
+// NewFilePositionCoordinate wraps a mysql.Position in a BinlogCoordinate.
+func NewFilePositionCoordinate(pos mysql.Position) BinlogCoordinate {
+	return BinlogCoordinate{
+		Type:         BinlogCoordinateFilePosition,
+		FilePosition: pos,
+	}
+}
+
+// resolvedType returns the effective type, treating the empty string as
+// file/position for backwards compatibility with older serialized state and
+// zero-valued coordinates.
+func (c BinlogCoordinate) resolvedType() BinlogCoordinateType {
+	if c.Type == "" {
+		return BinlogCoordinateFilePosition
+	}
+	return c.Type
+}
+
+// IsFilePosition reports whether this coordinate is a file/position coordinate.
+func (c BinlogCoordinate) IsFilePosition() bool {
+	return c.resolvedType() == BinlogCoordinateFilePosition
+}
+
+// Position returns the underlying mysql.Position.
+//
+// It is valid to call this only for file/position coordinates. This accessor
+// exists to keep the migration incremental: consumers that still require a
+// concrete mysql.Position can obtain it here while the surrounding plumbing is
+// converted to BinlogCoordinate.
+func (c BinlogCoordinate) Position() mysql.Position {
+	return c.FilePosition
+}
+
+// IsZero reports whether the coordinate carries no meaningful position. This
+// mirrors the previous use of an empty mysql.Position as the "unset" sentinel.
+func (c BinlogCoordinate) IsZero() bool {
+	switch c.resolvedType() {
+	case BinlogCoordinateFilePosition:
+		return c.FilePosition == (mysql.Position{})
+	default:
+		return false
+	}
+}
+
+// Compare orders two coordinates of the same type.
+//
+// It returns -1, 0 or 1 following the semantics of mysql.Position.Compare for
+// file/position coordinates. Comparing coordinates of differing types is a
+// programmer error and panics, because there is no meaningful total ordering
+// across representations. Callers that may receive mixed types should branch on
+// the coordinate type first.
+func (c BinlogCoordinate) Compare(other BinlogCoordinate) int {
+	if c.resolvedType() != other.resolvedType() {
+		panic(fmt.Sprintf(
+			"cannot compare binlog coordinates of different types: %q vs %q",
+			c.resolvedType(), other.resolvedType(),
+		))
+	}
+
+	switch c.resolvedType() {
+	case BinlogCoordinateFilePosition:
+		return c.FilePosition.Compare(other.FilePosition)
+	default:
+		panic(fmt.Sprintf("comparison not implemented for binlog coordinate type %q", c.resolvedType()))
+	}
+}
+
+// String returns a human readable representation for logs and status output.
+func (c BinlogCoordinate) String() string {
+	switch c.resolvedType() {
+	case BinlogCoordinateFilePosition:
+		return fmt.Sprintf("%s:%d", c.FilePosition.Name, c.FilePosition.Pos)
+	default:
+		return fmt.Sprintf("<unknown binlog coordinate type %q>", c.resolvedType())
+	}
+}
+
+// serializedBinlogCoordinate is the on-disk / on-wire shape of a
+// BinlogCoordinate. It is deliberately explicit and self-describing so that a
+// future GTID variant can be added as additional fields without breaking
+// existing readers.
+type serializedBinlogCoordinate struct {
+	Type         BinlogCoordinateType `json:"Type"`
+	FilePosition *mysql.Position      `json:"FilePosition,omitempty"`
+}
+
+// MarshalJSON implements json.Marshaler.
+func (c BinlogCoordinate) MarshalJSON() ([]byte, error) {
+	out := serializedBinlogCoordinate{Type: c.resolvedType()}
+
+	switch c.resolvedType() {
+	case BinlogCoordinateFilePosition:
+		pos := c.FilePosition
+		out.FilePosition = &pos
+	default:
+		return nil, fmt.Errorf("cannot marshal binlog coordinate of type %q", c.resolvedType())
+	}
+
+	return json.Marshal(out)
+}
+
+// UnmarshalJSON implements json.Unmarshaler.
+//
+// It accepts both the new self-describing form and, for defensive
+// compatibility, a bare mysql.Position object of the form {"Name":...,"Pos":...}.
+func (c *BinlogCoordinate) UnmarshalJSON(data []byte) error {
+	var typed serializedBinlogCoordinate
+	if err := json.Unmarshal(data, &typed); err == nil && typed.Type != "" {
+		c.Type = typed.Type
+		switch typed.Type {
+		case BinlogCoordinateFilePosition:
+			if typed.FilePosition != nil {
+				c.FilePosition = *typed.FilePosition
+			} else {
+				c.FilePosition = mysql.Position{}
+			}
+			return nil
+		default:
+			return fmt.Errorf("cannot unmarshal binlog coordinate of type %q", typed.Type)
+		}
+	}
+
+	// Fall back to a bare mysql.Position for older/simpler payloads.
+	var pos mysql.Position
+	if err := json.Unmarshal(data, &pos); err != nil {
+		return err
+	}
+	c.Type = BinlogCoordinateFilePosition
+	c.FilePosition = pos
+	return nil
+}

binlog_streamer.go

@@ -340,6 +340,12 @@ func (s *BinlogStreamer) GetLastStreamedBinlogPosition() mysql.Position {
 	return s.lastStreamedBinlogPosition
 }
 
+// GetLastStreamedBinlogCoordinate is the coordinate-typed counterpart of
+// GetLastStreamedBinlogPosition.
+func (s *BinlogStreamer) GetLastStreamedBinlogCoordinate() BinlogCoordinate {
+	return NewFilePositionCoordinate(s.lastStreamedBinlogPosition)
+}
+
 func (s *BinlogStreamer) IsAlmostCaughtUp() bool {
 	return time.Now().Sub(s.lastProcessedEventTime) < caughtUpThreshold
 }

dml_events.go

@@ -79,6 +79,11 @@ type DMLEvent interface {
 	PaginationKey() (string, error)
 	BinlogPosition() mysql.Position
 	ResumableBinlogPosition() mysql.Position
+	// BinlogCoordinate and ResumableBinlogCoordinate are the coordinate-typed
+	// counterparts of BinlogPosition and ResumableBinlogPosition. They are the
+	// forward-looking API used while Ghostferry migrates off raw mysql.Position.
+	BinlogCoordinate() BinlogCoordinate
+	ResumableBinlogCoordinate() BinlogCoordinate
 	Annotation() (string, error)
 	Timestamp() time.Time
 }
@@ -112,6 +117,14 @@ func (e *DMLEventBase) ResumableBinlogPosition() mysql.Position {
 	return e.resumablePos
 }
 
+func (e *DMLEventBase) BinlogCoordinate() BinlogCoordinate {
+	return NewFilePositionCoordinate(e.pos)
+}
+
+func (e *DMLEventBase) ResumableBinlogCoordinate() BinlogCoordinate {
+	return NewFilePositionCoordinate(e.resumablePos)
+}
+
 // Annotation will return the first prefixed comment on the SQL string,
 // or an error if the query attribute of the DMLEvent is not set
 func (e *DMLEventBase) Annotation() (string, error) {
@@ -504,10 +517,10 @@ func Int64Value(value interface{}) (int64, bool) {
 //
 // This is specifically mentioned in the the below link:
 //
-//    When BINARY values are stored, they are right-padded with the pad value
-//    to the specified length. The pad value is 0x00 (the zero byte). Values
-//    are right-padded with 0x00 for inserts, and no trailing bytes are removed
-//    for retrievals.
+//	When BINARY values are stored, they are right-padded with the pad value
+//	to the specified length. The pad value is 0x00 (the zero byte). Values
+//	are right-padded with 0x00 for inserts, and no trailing bytes are removed
+//	for retrievals.
 //
 // ref: https://dev.mysql.com/doc/refman/5.7/en/binary-varbinary.html
 func appendEscapedString(buffer []byte, value string, rightPadToLengthWithZeroBytes int) []byte {

state_tracker.go

@@ -49,7 +49,7 @@ func (s *SerializableState) MarshalJSON() ([]byte, error) {
 		LastSuccessfulPaginationKeys map[string]json.RawMessage
 		*Alias
 	}{
-		Alias: (*Alias)(s),
+		Alias:                        (*Alias)(s),
 		LastSuccessfulPaginationKeys: make(map[string]json.RawMessage),
 	}
 
@@ -106,6 +106,13 @@ func (s *SerializableState) MinSourceBinlogPosition() mysql.Position {
 	}
 }
 
+// MinSourceBinlogCoordinate returns the safe source resume coordinate. It is
+// the coordinate-typed counterpart of MinSourceBinlogPosition and currently
+// delegates to it, wrapping the result as a file/position coordinate.
+func (s *SerializableState) MinSourceBinlogCoordinate() BinlogCoordinate {
+	return NewFilePositionCoordinate(s.MinSourceBinlogPosition())
+}
+
 // For tracking the speed of the copy
 type PaginationKeyPositionLog struct {
 	Position float64
@@ -193,6 +200,46 @@ func (s *StateTracker) UpdateLastResumableBinlogPositionForTargetVerifier(pos my
 	s.lastStoredBinlogPositionForTargetVerifier = pos
 }
 
+// Coordinate-based accessors and mutators.
+//
+// These are the forward-looking API. They currently delegate to the
+// file/position storage above so behavior is unchanged, but they let callers be
+// written in terms of BinlogCoordinate ahead of introducing a GTID coordinate
+// mode.
+
+func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinate(coord BinlogCoordinate) {
+	s.UpdateLastResumableSourceBinlogPosition(coord.Position())
+}
+
+func (s *StateTracker) UpdateLastResumableSourceBinlogCoordinateForInlineVerifier(coord BinlogCoordinate) {
+	s.UpdateLastResumableSourceBinlogPositionForInlineVerifier(coord.Position())
+}
+
+func (s *StateTracker) UpdateLastResumableBinlogCoordinateForTargetVerifier(coord BinlogCoordinate) {
+	s.UpdateLastResumableBinlogPositionForTargetVerifier(coord.Position())
+}
+
+func (s *StateTracker) LastResumableSourceBinlogCoordinate() BinlogCoordinate {
+	s.BinlogRWMutex.RLock()
+	defer s.BinlogRWMutex.RUnlock()
+
+	return NewFilePositionCoordinate(s.lastWrittenBinlogPosition)
+}
+
+func (s *StateTracker) LastResumableSourceBinlogCoordinateForInlineVerifier() BinlogCoordinate {
+	s.BinlogRWMutex.RLock()
+	defer s.BinlogRWMutex.RUnlock()
+
+	return NewFilePositionCoordinate(s.lastStoredBinlogPositionForInlineVerifier)
+}
+
+func (s *StateTracker) LastResumableBinlogCoordinateForTargetVerifier() BinlogCoordinate {
+	s.BinlogRWMutex.RLock()
+	defer s.BinlogRWMutex.RUnlock()
+
+	return NewFilePositionCoordinate(s.lastStoredBinlogPositionForTargetVerifier)
+}
+
 func (s *StateTracker) UpdateLastSuccessfulPaginationKey(table string, paginationKey PaginationKey, rowStats RowStats) {
 	s.CopyRWMutex.Lock()
 	defer s.CopyRWMutex.Unlock()

test/go/binlog_coordinate_test.go

@@ -0,0 +1,78 @@
+package test
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/Shopify/ghostferry"
+	"github.com/go-mysql-org/go-mysql/mysql"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestBinlogCoordinate_FilePositionBasics(t *testing.T) {
+	pos := mysql.Position{Name: "mysql-bin.000123", Pos: 456}
+	coord := ghostferry.NewFilePositionCoordinate(pos)
+
+	assert.True(t, coord.IsFilePosition())
+	assert.False(t, coord.IsZero())
+	assert.Equal(t, pos, coord.Position())
+	assert.Equal(t, "mysql-bin.000123:456", coord.String())
+}
+
+func TestBinlogCoordinate_ZeroValueIsFilePosition(t *testing.T) {
+	var coord ghostferry.BinlogCoordinate
+
+	assert.True(t, coord.IsFilePosition())
+	assert.True(t, coord.IsZero())
+	assert.Equal(t, mysql.Position{}, coord.Position())
+}
+
+func TestBinlogCoordinate_Compare(t *testing.T) {
+	a := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 10})
+	b := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000001", Pos: 20})
+	c := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000002", Pos: 5})
+
+	assert.Equal(t, -1, a.Compare(b))
+	assert.Equal(t, 1, b.Compare(a))
+	assert.Equal(t, 0, a.Compare(a))
+	assert.Equal(t, -1, b.Compare(c))
+}
+
+func TestBinlogCoordinate_JSONRoundTrip(t *testing.T) {
+	coord := ghostferry.NewFilePositionCoordinate(mysql.Position{Name: "mysql-bin.000777", Pos: 999})
+
+	data, err := json.Marshal(coord)
+	require.NoError(t, err)
+
+	var decoded ghostferry.BinlogCoordinate
+	require.NoError(t, json.Unmarshal(data, &decoded))
+
+	assert.True(t, decoded.IsFilePosition())
+	assert.Equal(t, coord.Position(), decoded.Position())
+}
+
+// TestBinlogCoordinate_UnmarshalBareMysqlPosition guards the backwards
+// compatible decoding path: a bare {"Name":...,"Pos":...} object (the shape an
+// older mysql.Position produced) must decode into a file/position coordinate.
+func TestBinlogCoordinate_UnmarshalBareMysqlPosition(t *testing.T) {
+	raw := []byte(`{"Name":"mysql-bin.000042","Pos":314}`)
+
+	var decoded ghostferry.BinlogCoordinate
+	require.NoError(t, json.Unmarshal(raw, &decoded))
+
+	assert.True(t, decoded.IsFilePosition())
+	assert.Equal(t, "mysql-bin.000042", decoded.Position().Name)
+	assert.Equal(t, uint32(314), decoded.Position().Pos)
+}
+
+func TestBinlogCoordinate_UnmarshalTypedForm(t *testing.T) {
+	raw := []byte(`{"Type":"file_position","FilePosition":{"Name":"mysql-bin.000050","Pos":700}}`)
+
+	var decoded ghostferry.BinlogCoordinate
+	require.NoError(t, json.Unmarshal(raw, &decoded))
+
+	assert.True(t, decoded.IsFilePosition())
+	assert.Equal(t, "mysql-bin.000050", decoded.Position().Name)
+	assert.Equal(t, uint32(700), decoded.Position().Pos)
+}

test/go/binlog_writer_test.go

@@ -24,8 +24,14 @@ func (e *stubDMLEvent) NewValues() ghostferry.RowData           { return nil }
 func (e *stubDMLEvent) PaginationKey() (string, error)          { return "", nil }
 func (e *stubDMLEvent) BinlogPosition() mysql.Position          { return mysql.Position{} }
 func (e *stubDMLEvent) ResumableBinlogPosition() mysql.Position { return mysql.Position{} }
-func (e *stubDMLEvent) Annotation() (string, error)             { return "", nil }
-func (e *stubDMLEvent) Timestamp() time.Time                    { return time.Time{} }
+func (e *stubDMLEvent) BinlogCoordinate() ghostferry.BinlogCoordinate {
+	return ghostferry.NewFilePositionCoordinate(mysql.Position{})
+}
+func (e *stubDMLEvent) ResumableBinlogCoordinate() ghostferry.BinlogCoordinate {
+	return ghostferry.NewFilePositionCoordinate(mysql.Position{})
+}
+func (e *stubDMLEvent) Annotation() (string, error) { return "", nil }
+func (e *stubDMLEvent) Timestamp() time.Time        { return time.Time{} }
 
 // TestBinlogWriterBufferBinlogEventsBeforeRun verifies that BufferBinlogEvents
 // does not block when called before Run() has started in its own goroutine.