From 5738240280a90bc103ac0dcb3f6e50c2c23b0896 Mon Sep 17 00:00:00 2001 From: Omar Ramos Date: Mon, 3 Aug 2026 13:17:08 -0700 Subject: [PATCH 1/2] Add opt-in auto re-sync on schema incompatibility IsBinlogsExpirationError drops the saved position and lets the next iteration run a historical sync, so the connection recovers on its own. The schema-incompatibility path added in #89 returns an error instead, so a deploy that changes the layout of a replicated table stalls the connection until someone triggers a historical re-sync by hand. Until they do, every sync retries from the same position and fails on the same binlog event. Because a Read() error returns out of the whole Sync handler rather than skipping the affected table, one table in this state stalls every table in the connection. Unlike expired binlogs the operator has a real choice here, since the data is still readable once the projection is rebuilt, and an automatic historical sync costs monthly active rows. So this is opt-in via a new auto_resync_on_schema_change configuration field, defaulting to false, which preserves the existing behaviour exactly. With it enabled the cursor is reset and the stream recovers unattended. The cursor carries SCHEMA_INCOMPATIBILITY_ERROR rather than reusing BINLOG_EXPIRATION_ERROR, so the two causes stay distinguishable downstream and an unexpected historical sync remains attributable. Co-Authored-By: Claude Opus 5 (1M context) --- .../server/handlers/configuration_form.go | 13 ++++ cmd/internal/server/types.go | 8 +++ lib/connect_client.go | 26 ++++++- lib/connect_client_test.go | 67 ++++++++++++++++++- lib/planetscale_source.go | 5 ++ lib/types.go | 8 +++ 6 files changed, 125 insertions(+), 2 deletions(-) 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/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 } From 7a229908e21df79d7c428e474eb79c00bacda97f Mon Sep 17 00:00:00 2001 From: Omar Ramos Date: Tue, 4 Aug 2026 18:38:53 -0700 Subject: [PATCH 2/2] Cover the auto-resync flag from form field to parsed source The configuration form declares a field name and SourceFromRequest reads a map key. Nothing tied the two together, so if they ever drifted the toggle would fail silently: an operator enables it, the connector keeps the old behaviour, and no error is raised anywhere. Assert the round trip instead of the two literals independently -- read the field name out of the form, feed it through SourceFromRequest, and check the parsed value. Also covers the absent-key default staying false, so the pre-existing behaviour is pinned, and a non-boolean value being rejected rather than silently ignored. Verified the test fails when the parser's key is changed, so it guards the drift it claims to. Co-Authored-By: Claude Opus 5 (1M context) --- cmd/internal/server/server_test.go | 53 ++++++++++++++++++++++++++++++ 1 file changed, 53 insertions(+) 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") +}