From 713bb6afcc798a4d75b7e3641aa9bf0d126e1c6e Mon Sep 17 00:00:00 2001 From: bbernays Date: Wed, 15 Jul 2026 11:50:11 -0400 Subject: [PATCH 1/4] expose error classifier --- scheduler/queue/scheduler.go | 8 ++ scheduler/queue/worker.go | 12 +- scheduler/resolvers/error_classifier_test.go | 125 +++++++++++++++++++ scheduler/resolvers/resolvers.go | 61 ++++++--- scheduler/resolvers/resolvers_test.go | 4 +- scheduler/scheduler.go | 25 ++++ scheduler/scheduler_dfs.go | 7 +- scheduler/scheduler_error_classifier_test.go | 94 ++++++++++++++ scheduler/scheduler_shuffle_queue.go | 1 + schema/error_classifier.go | 54 ++++++++ 10 files changed, 369 insertions(+), 22 deletions(-) create mode 100644 scheduler/resolvers/error_classifier_test.go create mode 100644 scheduler/scheduler_error_classifier_test.go create mode 100644 schema/error_classifier.go diff --git a/scheduler/queue/scheduler.go b/scheduler/queue/scheduler.go index 0f830da76d..1a0241d1e7 100644 --- a/scheduler/queue/scheduler.go +++ b/scheduler/queue/scheduler.go @@ -33,6 +33,7 @@ type Scheduler struct { metrics *metrics.Metrics invocationID string seed int64 + errorClassifier schema.ErrorClassifier } type Option func(*Scheduler) @@ -61,6 +62,12 @@ func WithInvocationID(invocationID string) Option { } } +func WithErrorClassifier(classifier schema.ErrorClassifier) Option { + return func(d *Scheduler) { + d.errorClassifier = classifier + } +} + func NewShuffleQueueScheduler(logger zerolog.Logger, m *metrics.Metrics, seed int64, opts ...Option) *Scheduler { scheduler := &Scheduler{ logger: logger, @@ -104,6 +111,7 @@ func (d *Scheduler) Sync(ctx context.Context, tableClients []WorkUnit, resolvedR d.deterministicCQID, d.metrics, msgChan, + d.errorClassifier, ).work(ctx, activeWorkSignal) return nil }) diff --git a/scheduler/queue/worker.go b/scheduler/queue/worker.go index bd2c1485fd..39970244fa 100644 --- a/scheduler/queue/worker.go +++ b/scheduler/queue/worker.go @@ -32,7 +32,8 @@ type worker struct { deterministicCQID bool metrics *metrics.Metrics // message channel for sending SyncError messages - msgChan chan<- message.SyncMessage + msgChan chan<- message.SyncMessage + errorClassifier schema.ErrorClassifier } func (w *worker) work(ctx context.Context, activeWorkSignal *activeWorkSignal) { @@ -57,6 +58,7 @@ func newWorker( deterministicCQID bool, m *metrics.Metrics, msgChan chan<- message.SyncMessage, + errorClassifier schema.ErrorClassifier, ) *worker { return &worker{ jobs: jobs, @@ -68,6 +70,7 @@ func newWorker( invocationID: invocationID, metrics: m, msgChan: msgChan, + errorClassifier: errorClassifier, } } @@ -113,6 +116,11 @@ func (w *worker) resolveTable(ctx context.Context, table *schema.Table, client s close(res) }() if err := table.Resolver(ctx, client, parent, res); err != nil { + event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhaseTableResolver} + if w.errorClassifier.Suppress(ctx, err, event) { + logger.Debug().Err(err).Msg("table resolver finished with error (suppressed)") + return + } logger.Error().Err(err).Msg("table resolver finished with error") w.metrics.AddErrors(ctx, 1, selector) // Send SyncError message @@ -155,7 +163,7 @@ func (w *worker) resolveResource(ctx context.Context, table *schema.Table, clien wg.Add(1) go func() { defer wg.Done() - resolvedResources := resolvers.ResolveResourcesChunk(ctx, w.logger, w.metrics, table, client, parent, chunks[i], w.caser) + resolvedResources := resolvers.ResolveResourcesChunk(ctx, w.logger, w.metrics, table, client, parent, chunks[i], w.caser, w.errorClassifier) for _, resolvedResource := range resolvedResources { if err := resolvedResource.CalculateCQID(w.deterministicCQID); err != nil { w.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with primary key calculation error") diff --git a/scheduler/resolvers/error_classifier_test.go b/scheduler/resolvers/error_classifier_test.go new file mode 100644 index 0000000000..9bb7115ec5 --- /dev/null +++ b/scheduler/resolvers/error_classifier_test.go @@ -0,0 +1,125 @@ +package resolvers + +import ( + "context" + "errors" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/cloudquery/plugin-sdk/v4/caser" + "github.com/cloudquery/plugin-sdk/v4/scheduler/metrics" + "github.com/cloudquery/plugin-sdk/v4/schema" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +var errResolverBoom = errors.New("resolver boom") + +// TestResolveResourcesChunk_ErrorClassifier verifies that the error classifier controls +// whether resolver errors at each phase are counted as errors, and that it receives an +// ErrorEvent describing the phase that failed. +func TestResolveResourcesChunk_ErrorClassifier(t *testing.T) { + for _, tc := range []struct { + name string + table func() *schema.Table + wantPhase schema.ErrorPhase + wantColumn string + }{ + { + name: "column resolver", + table: func() *schema.Table { + return &schema.Table{ + Name: "test_table", + Columns: []schema.Column{ + { + Name: "test_column", + Type: arrow.PrimitiveTypes.Int64, + Resolver: func(_ context.Context, _ schema.ClientMeta, _ *schema.Resource, _ schema.Column) error { + return errResolverBoom + }, + }, + }, + } + }, + wantPhase: schema.ErrorPhaseColumnResolver, + wantColumn: "test_column", + }, + { + name: "pre resource chunk resolver", + table: func() *schema.Table { + return &schema.Table{ + Name: "test_table", + PreResourceChunkResolver: &schema.RowsChunkResolver{ + ChunkSize: 10, + RowsResolver: func(_ context.Context, _ schema.ClientMeta, _ []*schema.Resource) error { + return errResolverBoom + }, + }, + Columns: []schema.Column{{Name: "test_column", Type: arrow.PrimitiveTypes.Int64}}, + } + }, + wantPhase: schema.ErrorPhasePreResourceChunkResolver, + }, + { + name: "pre resource resolver", + table: func() *schema.Table { + return &schema.Table{ + Name: "test_table", + PreResourceResolver: func(_ context.Context, _ schema.ClientMeta, _ *schema.Resource) error { + return errResolverBoom + }, + Columns: []schema.Column{{Name: "test_column", Type: arrow.PrimitiveTypes.Int64}}, + } + }, + wantPhase: schema.ErrorPhasePreResourceResolver, + }, + { + name: "post resource resolver", + table: func() *schema.Table { + return &schema.Table{ + Name: "test_table", + PostResourceResolver: func(_ context.Context, _ schema.ClientMeta, _ *schema.Resource) error { + return errResolverBoom + }, + Columns: []schema.Column{{Name: "test_column", Type: arrow.PrimitiveTypes.Int64}}, + } + }, + wantPhase: schema.ErrorPhasePostResourceResolver, + }, + } { + t.Run(tc.name, func(t *testing.T) { + client := testClient{} + chunk := []any{0} + logger := zerolog.New(zerolog.NewTestWriter(t)) + + t.Run("raised without classifier", func(t *testing.T) { + table := tc.table() + m := metrics.NewMetrics() + m.InitWithClients(table, []schema.ClientMeta{client}) + ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), nil) + require.Equal(t, uint64(1), m.GetErrors(m.NewSelector(client.ID(), table.Name))) + }) + + t.Run("suppressed by classifier", func(t *testing.T) { + table := tc.table() + m := metrics.NewMetrics() + m.InitWithClients(table, []schema.ClientMeta{client}) + var gotEvent schema.ErrorEvent + classifier := func(_ context.Context, _ error, event schema.ErrorEvent) bool { + gotEvent = event + return true + } + ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), classifier) + require.Equal(t, uint64(0), m.GetErrors(m.NewSelector(client.ID(), table.Name)), "suppressed error should not be counted") + require.Equal(t, tc.wantPhase, gotEvent.Phase) + require.Equal(t, table, gotEvent.Table) + if tc.wantColumn != "" { + require.NotNil(t, gotEvent.Column) + require.Equal(t, tc.wantColumn, gotEvent.Column.Name) + } else { + require.Nil(t, gotEvent.Column) + } + }) + }) + } +} diff --git a/scheduler/resolvers/resolvers.go b/scheduler/resolvers/resolvers.go index 17718658eb..9a514ce7d7 100644 --- a/scheduler/resolvers/resolvers.go +++ b/scheduler/resolvers/resolvers.go @@ -14,7 +14,7 @@ import ( "github.com/thoas/go-funk" ) -func resolveColumn(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, selector metrics.Selector, client schema.ClientMeta, resource *schema.Resource, column schema.Column, c *caser.Caser) { +func resolveColumn(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, selector metrics.Selector, client schema.ClientMeta, resource *schema.Resource, column schema.Column, c *caser.Caser, classifier schema.ErrorClassifier) { columnStartTime := time.Now() defer func() { if err := recover(); err != nil { @@ -29,25 +29,32 @@ func resolveColumn(ctx context.Context, logger zerolog.Logger, m *metrics.Metric } }() + handleErr := func(err error) { + event := schema.ErrorEvent{Table: resource.Table, Client: client, Phase: schema.ErrorPhaseColumnResolver, Column: &column} + if classifier.Suppress(ctx, err, event) { + logger.Debug().Str("column", column.Name).Err(err).Msg("column resolver finished with error (suppressed)") + return + } + logger.Error().Err(err).Msg("column resolver finished with error") + m.AddErrors(ctx, 1, selector) + } + if column.Resolver != nil { if err := column.Resolver(ctx, client, resource, column); err != nil { - logger.Error().Err(err).Msg("column resolver finished with error") - m.AddErrors(ctx, 1, selector) + handleErr(err) } } else { // base use case: try to get column with CamelCase name v := funk.Get(resource.GetItem(), c.ToPascal(column.Name), funk.WithAllowZero()) if v != nil { - err := resource.Set(column.Name, v) - if err != nil { - logger.Error().Err(err).Msg("column resolver finished with error") - m.AddErrors(ctx, 1, selector) + if err := resource.Set(column.Name, v); err != nil { + handleErr(err) } } } } -func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, table *schema.Table, client schema.ClientMeta, parent *schema.Resource, chunk []any, c *caser.Caser) []*schema.Resource { +func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, table *schema.Table, client schema.ClientMeta, parent *schema.Resource, chunk []any, c *caser.Caser, classifier schema.ErrorClassifier) []*schema.Resource { ctx, cancel := context.WithTimeout(ctx, 10*time.Minute) defer cancel() @@ -72,8 +79,13 @@ func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metric if table.PreResourceChunkResolver != nil { if err := table.PreResourceChunkResolver.RowsResolver(ctx, client, resources); err != nil { - tableLogger.Error().Stack().Err(err).Msg("pre resource chunk resolver finished with error") - m.AddErrors(ctx, 1, selector) + event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhasePreResourceChunkResolver} + if classifier.Suppress(ctx, err, event) { + tableLogger.Debug().Err(err).Msg("pre resource chunk resolver finished with error (suppressed)") + } else { + tableLogger.Error().Stack().Err(err).Msg("pre resource chunk resolver finished with error") + m.AddErrors(ctx, 1, selector) + } return nil } } @@ -82,14 +94,24 @@ func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metric filtered := resources[:0] for _, resource := range resources { if err := table.PreResourceResolver(ctx, client, resource); err != nil { - if ctx.Err() != nil { + event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhasePreResourceResolver} + suppress := classifier.Suppress(ctx, err, event) + switch { + case suppress && ctx.Err() != nil: + tableLogger.Debug().Err(err).Msg("pre resource resolver failed, context cancelled (suppressed)") + return nil + case suppress: + tableLogger.Debug().Err(err).Msg("pre resource resolver failed (suppressed)") + continue + case ctx.Err() != nil: tableLogger.Error().Err(err).Msg("pre resource resolver failed, context cancelled") m.AddErrors(ctx, 1, selector) return nil + default: + tableLogger.Error().Err(err).Msg("pre resource resolver failed") + m.AddErrors(ctx, 1, selector) + continue } - tableLogger.Error().Err(err).Msg("pre resource resolver failed") - m.AddErrors(ctx, 1, selector) - continue } filtered = append(filtered, resource) } @@ -97,15 +119,20 @@ func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metric } for _, resource := range resources { for _, column := range table.Columns { - resolveColumn(ctx, tableLogger, m, selector, client, resource, column, c) + resolveColumn(ctx, tableLogger, m, selector, client, resource, column, c, classifier) } } if table.PostResourceResolver != nil { for _, resource := range resources { if err := table.PostResourceResolver(ctx, client, resource); err != nil { - tableLogger.Error().Stack().Err(err).Msg("post resource resolver finished with error") - m.AddErrors(ctx, 1, selector) + event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhasePostResourceResolver} + if classifier.Suppress(ctx, err, event) { + tableLogger.Debug().Err(err).Msg("post resource resolver finished with error (suppressed)") + } else { + tableLogger.Error().Stack().Err(err).Msg("post resource resolver finished with error") + m.AddErrors(ctx, 1, selector) + } } } } diff --git a/scheduler/resolvers/resolvers_test.go b/scheduler/resolvers/resolvers_test.go index 3af5c2b44f..45318456ea 100644 --- a/scheduler/resolvers/resolvers_test.go +++ b/scheduler/resolvers/resolvers_test.go @@ -76,7 +76,7 @@ func TestResolveResourcesChunk_PreResourceResolverPartialFailure(t *testing.T) { chunk := []any{0, 1, 2, 3, 4} logger := zerolog.New(zerolog.NewTestWriter(t)) - resources := ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New()) + resources := ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), nil) gotItems := make([]int, len(resources)) for i, r := range resources { @@ -131,7 +131,7 @@ func TestResolveResourcesChunk_PreResourceResolverContextCancelled(t *testing.T) chunk := []any{0, 1, 2, 3, 4} logger := zerolog.New(zerolog.NewTestWriter(t)) - resources := ResolveResourcesChunk(ctx, logger, m, table, client, nil, chunk, caser.New()) + resources := ResolveResourcesChunk(ctx, logger, m, table, client, nil, chunk, caser.New(), nil) require.Empty(t, resources, "cancelled chunk should not return resources") require.Equal(t, 2, calls, "resolver should not be called for resources after cancellation") diff --git a/scheduler/scheduler.go b/scheduler/scheduler.go index 8d17e08ffe..13913dbc80 100644 --- a/scheduler/scheduler.go +++ b/scheduler/scheduler.go @@ -37,6 +37,21 @@ const ( StrategyShuffleQueue ) +// Re-exported from the schema package for use with WithErrorClassifier. +type ( + ErrorClassifier = schema.ErrorClassifier + ErrorEvent = schema.ErrorEvent + ErrorPhase = schema.ErrorPhase +) + +const ( + ErrorPhaseTableResolver = schema.ErrorPhaseTableResolver + ErrorPhasePreResourceChunkResolver = schema.ErrorPhasePreResourceChunkResolver + ErrorPhasePreResourceResolver = schema.ErrorPhasePreResourceResolver + ErrorPhaseColumnResolver = schema.ErrorPhaseColumnResolver + ErrorPhasePostResourceResolver = schema.ErrorPhasePostResourceResolver +) + type Option func(*Scheduler) func WithLogger(logger zerolog.Logger) Option { @@ -45,6 +60,14 @@ func WithLogger(logger zerolog.Logger) Option { } } +// WithErrorClassifier sets the classifier consulted for every resolver error. See +// schema.ErrorClassifier. +func WithErrorClassifier(classifier ErrorClassifier) Option { + return func(s *Scheduler) { + s.errorClassifier = classifier + } +} + func WithConcurrency(concurrency int) Option { return func(s *Scheduler) { s.concurrency = concurrency @@ -122,6 +145,8 @@ type Scheduler struct { batchSettings *BatchSettings invocationID string + + errorClassifier schema.ErrorClassifier } type shard struct { diff --git a/scheduler/scheduler_dfs.go b/scheduler/scheduler_dfs.go index b90738ed69..308e4c69aa 100644 --- a/scheduler/scheduler_dfs.go +++ b/scheduler/scheduler_dfs.go @@ -128,6 +128,11 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c close(res) }() if err := table.Resolver(ctx, client, parent, res); err != nil { + event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhaseTableResolver} + if s.scheduler.errorClassifier.Suppress(ctx, err, event) { + logger.Debug().Err(err).Msg("table resolver finished with error (suppressed)") + return + } logger.Error().Err(err).Msg("table resolver finished with error") s.metrics.AddErrors(ctx, 1, selector) // Send SyncError message @@ -192,7 +197,7 @@ func (s *syncClient) resolveResourcesDfs(ctx context.Context, table *schema.Tabl defer resourceSem.Release(1) defer s.scheduler.resourceSem.Release(1) defer wg.Done() - resolvedResources := resolvers.ResolveResourcesChunk(ctx, s.logger, s.metrics, table, client, parent, chunks[i], s.scheduler.caser) + resolvedResources := resolvers.ResolveResourcesChunk(ctx, s.logger, s.metrics, table, client, parent, chunks[i], s.scheduler.caser, s.scheduler.errorClassifier) if len(resolvedResources) == 0 { return } diff --git a/scheduler/scheduler_error_classifier_test.go b/scheduler/scheduler_error_classifier_test.go new file mode 100644 index 0000000000..39c036c8f2 --- /dev/null +++ b/scheduler/scheduler_error_classifier_test.go @@ -0,0 +1,94 @@ +package scheduler + +import ( + "context" + "errors" + "sync" + "testing" + + "github.com/apache/arrow-go/v18/arrow" + "github.com/cloudquery/plugin-sdk/v4/message" + "github.com/cloudquery/plugin-sdk/v4/schema" + "github.com/rs/zerolog" + "github.com/stretchr/testify/require" +) + +var errTableResolverBoom = errors.New("table resolver boom") + +// TestSchedulerErrorClassifier verifies that a table resolver error is raised as a +// SyncError message by default, and suppressed (no SyncError) when the configured +// ErrorClassifier returns true for it. +func TestSchedulerErrorClassifier(t *testing.T) { + for _, strategy := range AllStrategies { + t.Run(strategy.String(), func(t *testing.T) { + table := &schema.Table{ + Name: "test_table", + Resolver: func(_ context.Context, _ schema.ClientMeta, _ *schema.Resource, _ chan<- any) error { + return errTableResolverBoom + }, + Columns: []schema.Column{ + {Name: "name", Type: arrow.BinaryTypes.String}, + }, + } + + t.Run("raised by default", func(t *testing.T) { + errs := syncErrors(t, strategy, table, nil) + require.Len(t, errs, 1, "table resolver error should be raised as a SyncError") + require.Equal(t, "test_table", errs[0].TableName) + require.Contains(t, errs[0].Error, errTableResolverBoom.Error()) + }) + + t.Run("suppressed by classifier", func(t *testing.T) { + var seen []schema.ErrorEvent + var mu sync.Mutex + classifier := func(_ context.Context, err error, event schema.ErrorEvent) bool { + mu.Lock() + seen = append(seen, event) + mu.Unlock() + return errors.Is(err, errTableResolverBoom) + } + errs := syncErrors(t, strategy, table, classifier) + require.Empty(t, errs, "suppressed table resolver error should not emit a SyncError") + + mu.Lock() + defer mu.Unlock() + require.Len(t, seen, 1) + require.Equal(t, schema.ErrorPhaseTableResolver, seen[0].Phase) + require.Equal(t, table, seen[0].Table) + require.NotNil(t, seen[0].Client) + }) + + t.Run("not suppressed when classifier returns false", func(t *testing.T) { + classifier := func(_ context.Context, _ error, _ schema.ErrorEvent) bool { + return false + } + errs := syncErrors(t, strategy, table, classifier) + require.Len(t, errs, 1, "a classifier returning false must not suppress the error") + }) + }) + } +} + +func syncErrors(t *testing.T, strategy Strategy, table *schema.Table, classifier schema.ErrorClassifier) []*message.SyncError { + t.Helper() + opts := []Option{ + WithLogger(zerolog.New(zerolog.NewTestWriter(t)).Level(zerolog.DebugLevel)), + WithStrategy(strategy), + } + if classifier != nil { + opts = append(opts, WithErrorClassifier(classifier)) + } + sc := NewScheduler(opts...) + + msgs := make(chan message.SyncMessage, 100) + require.NoError(t, sc.Sync(context.Background(), &testExecutionClient{}, schema.Tables{table}, msgs)) + close(msgs) + + var syncErrs []*message.SyncError + for msg := range msgs { + if e, ok := msg.(*message.SyncError); ok { + syncErrs = append(syncErrs, e) + } + } + return syncErrs +} diff --git a/scheduler/scheduler_shuffle_queue.go b/scheduler/scheduler_shuffle_queue.go index 0bc5867cd7..924b3e1eed 100644 --- a/scheduler/scheduler_shuffle_queue.go +++ b/scheduler/scheduler_shuffle_queue.go @@ -37,6 +37,7 @@ func (s *syncClient) syncShuffleQueue(ctx context.Context, resolvedResources cha queue.WithCaser(s.scheduler.caser), queue.WithDeterministicCQID(s.deterministicCQID), queue.WithInvocationID(s.invocationID), + queue.WithErrorClassifier(s.scheduler.errorClassifier), ) queueClients := make([]queue.WorkUnit, 0, len(tableClients)) for _, tc := range tableClients { diff --git a/schema/error_classifier.go b/schema/error_classifier.go new file mode 100644 index 0000000000..690984e488 --- /dev/null +++ b/schema/error_classifier.go @@ -0,0 +1,54 @@ +package schema + +import "context" + +// ErrorPhase identifies the resolver stage that produced an error. +type ErrorPhase int + +const ( + ErrorPhaseTableResolver ErrorPhase = iota + ErrorPhasePreResourceChunkResolver + ErrorPhasePreResourceResolver + ErrorPhaseColumnResolver + ErrorPhasePostResourceResolver +) + +func (p ErrorPhase) String() string { + switch p { + case ErrorPhaseTableResolver: + return "table_resolver" + case ErrorPhasePreResourceChunkResolver: + return "pre_resource_chunk_resolver" + case ErrorPhasePreResourceResolver: + return "pre_resource_resolver" + case ErrorPhaseColumnResolver: + return "column_resolver" + case ErrorPhasePostResourceResolver: + return "post_resource_resolver" + default: + return "unknown" + } +} + +// ErrorEvent describes the context in which a resolver error occurred. +type ErrorEvent struct { + Table *Table + Client ClientMeta + Phase ErrorPhase + // Column is set only when Phase == ErrorPhaseColumnResolver. + Column *Column +} + +// ErrorClassifier reports whether a resolver error should be suppressed rather than +// raised. Suppressed errors are logged at debug level and are not counted in error +// metrics or emitted as a SyncError message. A nil ErrorClassifier raises every error. +// It is not consulted for primary key calculation or validation errors. +type ErrorClassifier func(ctx context.Context, err error, event ErrorEvent) bool + +// Suppress is a nil-safe call: a nil ErrorClassifier returns false. +func (c ErrorClassifier) Suppress(ctx context.Context, err error, event ErrorEvent) bool { + if c == nil { + return false + } + return c(ctx, err, event) +} From c8f360b8aaf0eca1471b8fd9a5324980f37bd89c Mon Sep 17 00:00:00 2001 From: bbernays Date: Wed, 15 Jul 2026 15:09:31 -0400 Subject: [PATCH 2/4] wording --- scheduler/queue/worker.go | 2 +- scheduler/resolvers/resolvers.go | 6 +++--- scheduler/scheduler_dfs.go | 2 +- 3 files changed, 5 insertions(+), 5 deletions(-) diff --git a/scheduler/queue/worker.go b/scheduler/queue/worker.go index 39970244fa..71cc9ccb1d 100644 --- a/scheduler/queue/worker.go +++ b/scheduler/queue/worker.go @@ -118,7 +118,7 @@ func (w *worker) resolveTable(ctx context.Context, table *schema.Table, client s if err := table.Resolver(ctx, client, parent, res); err != nil { event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhaseTableResolver} if w.errorClassifier.Suppress(ctx, err, event) { - logger.Debug().Err(err).Msg("table resolver finished with error (suppressed)") + logger.Debug().Err(err).Msg("table resolver finished with suppressed error") return } logger.Error().Err(err).Msg("table resolver finished with error") diff --git a/scheduler/resolvers/resolvers.go b/scheduler/resolvers/resolvers.go index 9a514ce7d7..f7168dcc63 100644 --- a/scheduler/resolvers/resolvers.go +++ b/scheduler/resolvers/resolvers.go @@ -32,7 +32,7 @@ func resolveColumn(ctx context.Context, logger zerolog.Logger, m *metrics.Metric handleErr := func(err error) { event := schema.ErrorEvent{Table: resource.Table, Client: client, Phase: schema.ErrorPhaseColumnResolver, Column: &column} if classifier.Suppress(ctx, err, event) { - logger.Debug().Str("column", column.Name).Err(err).Msg("column resolver finished with error (suppressed)") + logger.Debug().Str("column", column.Name).Err(err).Msg("column resolver finished with suppressed error") return } logger.Error().Err(err).Msg("column resolver finished with error") @@ -81,7 +81,7 @@ func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metric if err := table.PreResourceChunkResolver.RowsResolver(ctx, client, resources); err != nil { event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhasePreResourceChunkResolver} if classifier.Suppress(ctx, err, event) { - tableLogger.Debug().Err(err).Msg("pre resource chunk resolver finished with error (suppressed)") + tableLogger.Debug().Err(err).Msg("pre resource chunk resolver finished with suppressed error") } else { tableLogger.Error().Stack().Err(err).Msg("pre resource chunk resolver finished with error") m.AddErrors(ctx, 1, selector) @@ -128,7 +128,7 @@ func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metric if err := table.PostResourceResolver(ctx, client, resource); err != nil { event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhasePostResourceResolver} if classifier.Suppress(ctx, err, event) { - tableLogger.Debug().Err(err).Msg("post resource resolver finished with error (suppressed)") + tableLogger.Debug().Err(err).Msg("post resource resolver finished with suppressed error") } else { tableLogger.Error().Stack().Err(err).Msg("post resource resolver finished with error") m.AddErrors(ctx, 1, selector) diff --git a/scheduler/scheduler_dfs.go b/scheduler/scheduler_dfs.go index 308e4c69aa..6ab9d0c34f 100644 --- a/scheduler/scheduler_dfs.go +++ b/scheduler/scheduler_dfs.go @@ -130,7 +130,7 @@ func (s *syncClient) resolveTableDfs(ctx context.Context, table *schema.Table, c if err := table.Resolver(ctx, client, parent, res); err != nil { event := schema.ErrorEvent{Table: table, Client: client, Phase: schema.ErrorPhaseTableResolver} if s.scheduler.errorClassifier.Suppress(ctx, err, event) { - logger.Debug().Err(err).Msg("table resolver finished with error (suppressed)") + logger.Debug().Err(err).Msg("table resolver finished with suppressed error") return } logger.Error().Err(err).Msg("table resolver finished with error") From 954d447f063fb12f64ca401b17adcf3b4c2387ce Mon Sep 17 00:00:00 2001 From: bbernays Date: Thu, 16 Jul 2026 08:29:50 -0400 Subject: [PATCH 3/4] ensure change is not breaking --- scheduler/queue/worker.go | 2 +- scheduler/resolvers/error_classifier_test.go | 4 ++-- scheduler/resolvers/resolvers.go | 8 +++++++- scheduler/resolvers/resolvers_test.go | 4 ++-- scheduler/scheduler_dfs.go | 2 +- 5 files changed, 13 insertions(+), 7 deletions(-) diff --git a/scheduler/queue/worker.go b/scheduler/queue/worker.go index 71cc9ccb1d..1ee06eea42 100644 --- a/scheduler/queue/worker.go +++ b/scheduler/queue/worker.go @@ -163,7 +163,7 @@ func (w *worker) resolveResource(ctx context.Context, table *schema.Table, clien wg.Add(1) go func() { defer wg.Done() - resolvedResources := resolvers.ResolveResourcesChunk(ctx, w.logger, w.metrics, table, client, parent, chunks[i], w.caser, w.errorClassifier) + resolvedResources := resolvers.ResolveResourcesChunkWithClassifier(ctx, w.logger, w.metrics, table, client, parent, chunks[i], w.caser, w.errorClassifier) for _, resolvedResource := range resolvedResources { if err := resolvedResource.CalculateCQID(w.deterministicCQID); err != nil { w.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with primary key calculation error") diff --git a/scheduler/resolvers/error_classifier_test.go b/scheduler/resolvers/error_classifier_test.go index 9bb7115ec5..533c7f993c 100644 --- a/scheduler/resolvers/error_classifier_test.go +++ b/scheduler/resolvers/error_classifier_test.go @@ -96,7 +96,7 @@ func TestResolveResourcesChunk_ErrorClassifier(t *testing.T) { table := tc.table() m := metrics.NewMetrics() m.InitWithClients(table, []schema.ClientMeta{client}) - ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), nil) + ResolveResourcesChunkWithClassifier(context.Background(), logger, m, table, client, nil, chunk, caser.New(), nil) require.Equal(t, uint64(1), m.GetErrors(m.NewSelector(client.ID(), table.Name))) }) @@ -109,7 +109,7 @@ func TestResolveResourcesChunk_ErrorClassifier(t *testing.T) { gotEvent = event return true } - ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), classifier) + ResolveResourcesChunkWithClassifier(context.Background(), logger, m, table, client, nil, chunk, caser.New(), classifier) require.Equal(t, uint64(0), m.GetErrors(m.NewSelector(client.ID(), table.Name)), "suppressed error should not be counted") require.Equal(t, tc.wantPhase, gotEvent.Phase) require.Equal(t, table, gotEvent.Table) diff --git a/scheduler/resolvers/resolvers.go b/scheduler/resolvers/resolvers.go index f7168dcc63..f5d3387530 100644 --- a/scheduler/resolvers/resolvers.go +++ b/scheduler/resolvers/resolvers.go @@ -54,7 +54,13 @@ func resolveColumn(ctx context.Context, logger zerolog.Logger, m *metrics.Metric } } -func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, table *schema.Table, client schema.ClientMeta, parent *schema.Resource, chunk []any, c *caser.Caser, classifier schema.ErrorClassifier) []*schema.Resource { +// Deprecated: use ResolveResourcesChunkWithClassifier. This retains the original +// signature and resolves with a nil classifier, so every error is raised. +func ResolveResourcesChunk(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, table *schema.Table, client schema.ClientMeta, parent *schema.Resource, chunk []any, c *caser.Caser) []*schema.Resource { + return ResolveResourcesChunkWithClassifier(ctx, logger, m, table, client, parent, chunk, c, nil) +} + +func ResolveResourcesChunkWithClassifier(ctx context.Context, logger zerolog.Logger, m *metrics.Metrics, table *schema.Table, client schema.ClientMeta, parent *schema.Resource, chunk []any, c *caser.Caser, classifier schema.ErrorClassifier) []*schema.Resource { ctx, cancel := context.WithTimeout(ctx, 10*time.Minute) defer cancel() diff --git a/scheduler/resolvers/resolvers_test.go b/scheduler/resolvers/resolvers_test.go index 45318456ea..3af5c2b44f 100644 --- a/scheduler/resolvers/resolvers_test.go +++ b/scheduler/resolvers/resolvers_test.go @@ -76,7 +76,7 @@ func TestResolveResourcesChunk_PreResourceResolverPartialFailure(t *testing.T) { chunk := []any{0, 1, 2, 3, 4} logger := zerolog.New(zerolog.NewTestWriter(t)) - resources := ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New(), nil) + resources := ResolveResourcesChunk(context.Background(), logger, m, table, client, nil, chunk, caser.New()) gotItems := make([]int, len(resources)) for i, r := range resources { @@ -131,7 +131,7 @@ func TestResolveResourcesChunk_PreResourceResolverContextCancelled(t *testing.T) chunk := []any{0, 1, 2, 3, 4} logger := zerolog.New(zerolog.NewTestWriter(t)) - resources := ResolveResourcesChunk(ctx, logger, m, table, client, nil, chunk, caser.New(), nil) + resources := ResolveResourcesChunk(ctx, logger, m, table, client, nil, chunk, caser.New()) require.Empty(t, resources, "cancelled chunk should not return resources") require.Equal(t, 2, calls, "resolver should not be called for resources after cancellation") diff --git a/scheduler/scheduler_dfs.go b/scheduler/scheduler_dfs.go index 6ab9d0c34f..b810bedb1c 100644 --- a/scheduler/scheduler_dfs.go +++ b/scheduler/scheduler_dfs.go @@ -197,7 +197,7 @@ func (s *syncClient) resolveResourcesDfs(ctx context.Context, table *schema.Tabl defer resourceSem.Release(1) defer s.scheduler.resourceSem.Release(1) defer wg.Done() - resolvedResources := resolvers.ResolveResourcesChunk(ctx, s.logger, s.metrics, table, client, parent, chunks[i], s.scheduler.caser, s.scheduler.errorClassifier) + resolvedResources := resolvers.ResolveResourcesChunkWithClassifier(ctx, s.logger, s.metrics, table, client, parent, chunks[i], s.scheduler.caser, s.scheduler.errorClassifier) if len(resolvedResources) == 0 { return } From befd490faea2fef55be2ebcda9e132fcc6871090 Mon Sep 17 00:00:00 2001 From: bbernays Date: Thu, 16 Jul 2026 08:34:48 -0400 Subject: [PATCH 4/4] Update error_classifier.go --- schema/error_classifier.go | 1 + 1 file changed, 1 insertion(+) diff --git a/schema/error_classifier.go b/schema/error_classifier.go index 690984e488..dae84b6b6e 100644 --- a/schema/error_classifier.go +++ b/schema/error_classifier.go @@ -43,6 +43,7 @@ type ErrorEvent struct { // raised. Suppressed errors are logged at debug level and are not counted in error // metrics or emitted as a SyncError message. A nil ErrorClassifier raises every error. // It is not consulted for primary key calculation or validation errors. +// The classifier may be invoked concurrently; implementations must be safe for concurrent use. type ErrorClassifier func(ctx context.Context, err error, event ErrorEvent) bool // Suppress is a nil-safe call: a nil ErrorClassifier returns false.