diff --git a/cmd/internal/server/handlers/configuration_form.go b/cmd/internal/server/handlers/configuration_form.go index 8ac8c89..9bedaf0 100644 --- a/cmd/internal/server/handlers/configuration_form.go +++ b/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", diff --git a/cmd/internal/server/server_test.go b/cmd/internal/server/server_test.go index 51b7745..ee165a5 100644 --- a/cmd/internal/server/server_test.go +++ b/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") +} diff --git a/cmd/internal/server/types.go b/cmd/internal/server/types.go index 087dde8..78c8ffc 100644 --- a/cmd/internal/server/types.go +++ b/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 } diff --git a/lib/connect_client.go b/lib/connect_client.go index 6190e12..655acf8 100644 --- a/lib/connect_client.go +++ b/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 diff --git a/lib/connect_client_test.go b/lib/connect_client_test.go index e6a5252..e30c6a0 100644 --- a/lib/connect_client_test.go +++ b/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) diff --git a/lib/planetscale_source.go b/lib/planetscale_source.go index 9ca386d..7dc7fa0 100644 --- a/lib/planetscale_source.go +++ b/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. diff --git a/lib/types.go b/lib/types.go index d068821..afdebf4 100644 --- a/lib/types.go +++ b/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 }