Add opt-in auto re-sync on schema incompatibility

cmd/internal/server/handlers/configuration_form.go

@@ -17,6 +17,7 @@ func (ConfigurationForm) Handle(ctx context.Context, _ *fivetransdk.Configuratio
 	passwordDesc := "Password to connect to your PlanetScale database"
 	tinyIntDesc := "Enable this setting to serialize tinyint(1) as boolean values"
 	useReplicaDesc := "Only set to true if your PlanetScale branch has a replica. PlanetScale Development branches do not have replicas."
+	autoResyncDesc := "When a schema change leaves the saved position unreadable, automatically reset the cursor and run a historical sync for the affected table instead of failing the sync and waiting for a manual re-sync. Defaults to false; enabling it trades additional monthly active rows for unattended recovery."
 	required := true
 	resp := &fivetransdk.ConfigurationFormResponse{
 		Fields: []*fivetransdk.FormField{
@@ -87,6 +88,18 @@ func (ConfigurationForm) Handle(ctx context.Context, _ *fivetransdk.Configuratio
 					},
 				},
 			},
+			{
+				Name:        "auto_resync_on_schema_change",
+				Label:       "Automatically re-sync after a schema change?",
+				Description: &autoResyncDesc,
+				Type: &fivetransdk.FormField_DropdownField{
+					DropdownField: &fivetransdk.DropdownField{
+						DropdownField: []string{
+							"true", "false",
+						},
+					},
+				},
+			},
 			{
 				Name:        "starting_gtids",
 				Label:       "JSON containing keyspace, shard, and starting GTIDs",

cmd/internal/server/server_test.go

@@ -1100,3 +1100,56 @@ func TestSchemaChecksCredentials(t *testing.T) {
 	})
 	assert.ErrorContains(t, err, "unable to connect to PlanetScale database")
 }
+
+// configuredRequestStub satisfies ConfiguredRequest for parser tests.
+type configuredRequestStub struct{ cfg map[string]string }
+
+func (c configuredRequestStub) GetConfiguration() map[string]string { return c.cfg }
+
+// withConnDetails adds the connection fields SourceFromRequest requires, so the
+// test asserts on the flag rather than tripping over unrelated validation.
+func withConnDetails(extra map[string]string) configuredRequestStub {
+	cfg := map[string]string{
+		"username": "u", "password": "p", "database": "d", "host": "h",
+	}
+	for k, v := range extra {
+		cfg[k] = v
+	}
+	return configuredRequestStub{cfg: cfg}
+}
+
+// The setup form declares a field name and SourceFromRequest reads a map key.
+// If those ever drift apart the toggle silently stops working: the operator sets
+// it, nothing happens, and no error is raised. Assert the round trip rather than
+// the two literals independently.
+func TestAutoResyncFlagRoundTripsFromConfigurationForm(t *testing.T) {
+	ctx := context.Background()
+	form, err := (handlers.ConfigurationForm{}).Handle(ctx, &fivetransdk.ConfigurationFormRequest{})
+	assert.NoError(t, err)
+
+	var fieldName string
+	for _, f := range form.Fields {
+		if f.Name == "auto_resync_on_schema_change" {
+			fieldName = f.Name
+			assert.NotNil(t, f.GetDropdownField(), "expected a dropdown, matching the other boolean fields")
+			assert.Equal(t, []string{"true", "false"}, f.GetDropdownField().DropdownField)
+		}
+	}
+	assert.NotEmpty(t, fieldName, "auto_resync_on_schema_change missing from the configuration form")
+
+	on, err := SourceFromRequest(withConnDetails(map[string]string{fieldName: "true"}))
+	assert.NoError(t, err)
+	assert.True(t, on.AutoResyncOnSchemaChange, "form field name does not match the key SourceFromRequest reads")
+
+	off, err := SourceFromRequest(withConnDetails(map[string]string{fieldName: "false"}))
+	assert.NoError(t, err)
+	assert.False(t, off.AutoResyncOnSchemaChange)
+
+	// Absent key must default to false, preserving the pre-existing behaviour.
+	def, err := SourceFromRequest(withConnDetails(nil))
+	assert.NoError(t, err)
+	assert.False(t, def.AutoResyncOnSchemaChange)
+
+	_, err = SourceFromRequest(withConnDetails(map[string]string{fieldName: "yes-please"}))
+	assert.Error(t, err, "non-boolean value should be rejected rather than silently ignored")
+}

cmd/internal/server/types.go

@@ -101,6 +101,14 @@ func SourceFromRequest(request ConfiguredRequest) (*lib.PlanetScaleSource, error
 		psc.UseReplica = b
 	}
 
+	if val, ok := configuration["auto_resync_on_schema_change"]; ok {
+		b, err := strconv.ParseBool(val)
+		if err != nil {
+			return nil, errors.New("auto_resync_on_schema_change is not a boolean")
+		}
+		psc.AutoResyncOnSchemaChange = b
+	}
+
 	if val, ok := configuration["starting_gtids"]; ok {
 		psc.StartingGtids = val
 	}

lib/connect_client.go

@@ -193,8 +193,32 @@ func (p connectClient) Read(ctx context.Context, logger DatabaseLogger, ps Plane
 						continue
 					}
 
+					// A schema change has left the saved position undecodable for
+					// this table. Recovery is the same as for expired binlogs --
+					// drop the position and let the next iteration run a
+					// historical sync -- but unlike expired binlogs the operator
+					// has a real choice here, because the data is still readable
+					// from the saved position once the projection is rebuilt.
+					// An automatic re-sync costs monthly active rows, so it is
+					// opt-in; by default the error is surfaced and the sync waits
+					// for an operator-triggered historical re-sync.
 					if IsVStreamSchemaIncompatibilityError(err) {
-						return currentSerializedCursor, errors.Wrapf(err, "PlanetScale VStream cannot continue incremental replication for table %s after a schema change; run a Fivetran historical re-sync for this connection to recover", tableName)
+						if !ps.AutoResyncOnSchemaChange {
+							return currentSerializedCursor, errors.Wrapf(err, "PlanetScale VStream cannot continue incremental replication for table %s after a schema change; run a Fivetran historical re-sync for this connection to recover, or enable auto_resync_on_schema_change to have the connector do it automatically", tableName)
+						}
+
+						logger.Warning(fmt.Sprintf("%sSchema changed incompatibly with the saved cursor position. Resetting cursor position to trigger historical sync", preamble))
+						currentPosition.Position = ""
+						currentPosition.LastKnownPk = nil
+
+						currentSerializedCursor, sErr = TableCursorToSerializedCursor(currentPosition)
+						if sErr != nil {
+							return currentSerializedCursor, errors.Wrap(sErr, "unable to serialize reset cursor after schema incompatibility")
+						}
+						currentSerializedCursor.SetSchemaIncompatibilityError(fmt.Sprintf("PlanetScale VStream cannot continue incremental replication for table %s after a schema change. Cursor reset to trigger historical sync. Original error: %v", tableName, err.Error()))
+
+						// Continue with historical sync instead of returning error
+						continue
 					}
 
 					return currentSerializedCursor, err

lib/connect_client_test.go

@@ -331,7 +331,71 @@ func TestRead_CanReturnNewCursorIfNewFound(t *testing.T) {
 	assert.Equal(t, 2, cc.syncFnInvokedCount)
 }
 
-func TestRead_ReturnsVStreamSchemaIncompatibilityErrors(t *testing.T) {
+func TestRead_SchemaIncompatibilityResetsCursor(t *testing.T) {
+	dbl := &dbLogger{}
+	ped := connectClient{}
+	tc := &psdbconnect.TableCursor{
+		Shard:    "-",
+		Position: "THIS_IS_A_SHARD_GTID",
+		Keyspace: "connect-test",
+	}
+	stopCursor := &psdbconnect.TableCursor{
+		Shard:    "-",
+		Position: "I_AM_THE_CURRENT_BINLOG_POSITION",
+		Keyspace: "connect-test",
+	}
+
+	// The second peek fails so the loop terminates once the cursor has been
+	// reset, letting the test inspect the reset cursor that gets handed back.
+	currentCursorRequests := 0
+	cc := clientConnectionMock{
+		syncFn: func(ctx context.Context, in *psdbconnect.SyncRequest, opts ...grpc.CallOption) (psdbconnect.Connect_SyncClient, error) {
+			if in.Cursor.Position == "current" {
+				currentCursorRequests++
+				if currentCursorRequests == 1 {
+					return &connectSyncClientMock{
+						syncResponses: []*psdbconnect.SyncResponse{{Cursor: stopCursor}},
+					}, nil
+				}
+				return nil, errors.New("peek failed after reset")
+			}
+			return nil, status.Error(codes.Unknown, vstreamColumnNotFoundErrorMessage)
+		},
+	}
+	ped.clientFn = func(ctx context.Context, ps PlanetScaleSource) (psdbconnect.ConnectClient, error) {
+		return &cc, nil
+	}
+	getKeyspaceTableColumnsFunc := func(ctx context.Context, keyspaceName string, tableName string) ([]MysqlColumn, error) {
+		return []MysqlColumn{
+			{Name: "id", Type: "bigint", IsPrimaryKey: true},
+			{Name: "before_col", Type: "varchar(64)", IsPrimaryKey: false},
+			{Name: "after_col", Type: "varchar(64)", IsPrimaryKey: false},
+		}, nil
+	}
+	mysqlClient := NewTestMysqlClient(getKeyspaceTableColumnsFunc)
+	ped.Mysql = &mysqlClient
+
+	source := PlanetScaleSource{Database: "connect-test", AutoResyncOnSchemaChange: true}
+	sc, err := ped.Read(context.Background(), dbl, source, "customers", []string{"id", "before_col", "after_col"}, tc, nil, nil, nil)
+	assert.ErrorContains(t, err, "peek failed after reset")
+	if assert.NotNil(t, sc) {
+		cursor, cErr := sc.SerializedCursorToTableCursor()
+		assert.NoError(t, cErr)
+		assert.Empty(t, cursor.Position)
+		assert.Nil(t, cursor.LastKnownPk)
+		if assert.NotNil(t, sc.ErrorCode) {
+			assert.Equal(t, "SCHEMA_INCOMPATIBILITY_ERROR", *sc.ErrorCode)
+		}
+		if assert.NotNil(t, sc.ErrorMessage) {
+			assert.Contains(t, *sc.ErrorMessage, "historical sync")
+		}
+	}
+	assert.Equal(t, 3, cc.syncFnInvokedCount)
+}
+
+// Without the opt-in the connector keeps the pre-existing contract: surface the
+// error with recovery guidance and leave the saved cursor untouched.
+func TestRead_SchemaIncompatibilityWithoutOptInReturnsError(t *testing.T) {
 	dbl := &dbLogger{}
 	ped := connectClient{}
 	tc := &psdbconnect.TableCursor{
@@ -375,6 +439,7 @@ func TestRead_ReturnsVStreamSchemaIncompatibilityErrors(t *testing.T) {
 	sc, err := ped.Read(context.Background(), dbl, PlanetScaleSource{Database: "connect-test"}, "customers", []string{"id", "before_col", "after_col"}, tc, nil, nil, nil)
 	assert.Error(t, err)
 	assert.Contains(t, err.Error(), "historical re-sync")
+	assert.Contains(t, err.Error(), "auto_resync_on_schema_change")
 	assert.True(t, IsVStreamSchemaIncompatibilityError(err))
 	esc, err := TableCursorToSerializedCursor(tc)
 	assert.NoError(t, err)

lib/planetscale_source.go

@@ -21,6 +21,11 @@ type PlanetScaleSource struct {
 	TreatTinyIntAsBoolean bool   `json:"treat_tiny_int_as_boolean"`
 	UseReplica            bool   `json:"use_replica"`
 	StartingGtids         string `json:"starting_gtids"`
+	// AutoResyncOnSchemaChange opts in to resetting the cursor and running a
+	// historical sync when a schema change leaves the saved position
+	// undecodable. Defaults to false, which keeps the existing behaviour of
+	// surfacing the error and waiting for an operator-triggered re-sync.
+	AutoResyncOnSchemaChange bool `json:"auto_resync_on_schema_change"`
 }
 
 // DSN returns a DataSource that mysql libraries can use to connect to a PlanetScale database.

lib/types.go

@@ -74,6 +74,14 @@ func (sc *SerializedCursor) SetBinlogExpirationError(errorMessage string) {
 	sc.ErrorMessage = &errorMessage
 }
 
+// SetSchemaIncompatibilityError sets a schema incompatibility error on the cursor.
+// Carries its own code so that a re-sync forced by a schema change can be told
+// apart from one forced by binlog expiration, even though both reset the cursor.
+func (sc *SerializedCursor) SetSchemaIncompatibilityError(errorMessage string) {
+	sc.ErrorCode = stringPtr("SCHEMA_INCOMPATIBILITY_ERROR")
+	sc.ErrorMessage = &errorMessage
+}
+
 func stringPtr(s string) *string {
 	return &s
 }