From 31d1b136ce4e5f94e9a83ce7bc3bb5b04596a994 Mon Sep 17 00:00:00 2001 From: meredith <4412188+mariduv@users.noreply.github.com> Date: Wed, 6 Aug 2025 10:33:45 -0500 Subject: [PATCH] [SCAN-81] Report ChunkUnit Panics (#4367) * Add common.RecoverWithHandler Allows panic recovery that still reports to Sentry, with caller-defined handling afterward, so there's no confusion about chained recover() or re-panicking. * Catch ChunkUnit panics and include them in report Keeps panics showing in Sentry, and includes panics in existing first-fatal-error behavior, but does not propagate panics further upward any longer. * Test SourceManager ChunkUnit panic reporting --- pkg/common/recover.go | 17 +++++++++++++++++ pkg/sources/source_manager.go | 29 ++++++++++++++++++++++++----- pkg/sources/source_manager_test.go | 23 ++++++++++++++++++----- 3 files changed, 59 insertions(+), 10 deletions(-) diff --git a/pkg/common/recover.go b/pkg/common/recover.go index 3fa294b83..5edbbde82 100644 --- a/pkg/common/recover.go +++ b/pkg/common/recover.go @@ -27,6 +27,23 @@ func Recover(ctx context.Context) { } } +// RecoverWithHandler handles panics and reports to Sentry, then turns control +// over to a provided function. This permits extra reporting in the same scope +// without re-panicking, as recover() clears the state after it's called. Does +// NOT block to flush sentry report. +func RecoverWithHandler(ctx context.Context, callback func(error)) { + if err := recover(); err != nil { + panicStack := string(debug.Stack()) + if eventID := sentry.CurrentHub().Recover(err); eventID != nil { + ctx.Logger().Info("panic captured", "event_id", *eventID) + } + ctx.Logger().Error(fmt.Errorf("panic"), panicStack, + "recover", err, + ) + callback(fmt.Errorf("panic: %e", err)) + } +} + // RecoverWithExit handles panics and reports to Sentry before exiting. func RecoverWithExit(ctx context.Context) { if err := recover(); err != nil { diff --git a/pkg/sources/source_manager.go b/pkg/sources/source_manager.go index 68dd57f77..562a93947 100644 --- a/pkg/sources/source_manager.go +++ b/pkg/sources/source_manager.go @@ -157,7 +157,13 @@ func (s *SourceManager) EnumerateAndScan(ctx context.Context, sourceName string, ctx := context.WithValues(ctx, "source_manager_worker_id", common.RandomID(5), ) - defer common.Recover(ctx) + defer common.RecoverWithHandler(ctx, func(err error) { + progress.ReportError(Fatal{err}) + select { + case s.firstErr <- err: + default: + } + }) defer cancel(nil) if err := s.run(ctx, source, progress, targets...); err != nil { select { @@ -458,10 +464,13 @@ func (s *SourceManager) enumerateWithUnits(ctx context.Context, source SourceUni // Produce units. func() { - // TODO: Catch panics and add to report. report.StartEnumerating(time.Now()) defer func() { report.EndEnumerating(time.Now()) }() ctx.Logger().V(2).Info("enumerating source with units") + defer common.RecoverWithHandler(ctx, func(err error) { + report.ReportError(Fatal{err}) + catchFirstFatal(Fatal{err}) + }) if err := source.Enumerate(ctx, reporter); err != nil { report.ReportError(Fatal{err}) catchFirstFatal(Fatal{err}) @@ -523,13 +532,17 @@ func (s *SourceManager) runWithUnits(ctx context.Context, source SourceUnitEnumC default: } } + // Produce units. go func() { - // TODO: Catch panics and add to report. report.StartEnumerating(time.Now()) defer func() { report.EndEnumerating(time.Now()) }() defer close(unitReporter.unitCh) ctx.Logger().V(2).Info("enumerating source") + defer common.RecoverWithHandler(ctx, func(err error) { + report.ReportError(Fatal{err}) + catchFirstFatal(Fatal{err}) + }) if err := source.Enumerate(ctx, unitReporter); err != nil { report.ReportError(Fatal{err}) catchFirstFatal(Fatal{err}) @@ -551,11 +564,14 @@ func (s *SourceManager) runWithUnits(ctx context.Context, source SourceUnitEnumC // Consume units and produce chunks. unitPool.Go(func() error { report.StartUnitChunking(unit, time.Now()) - // TODO: Catch panics and add to report. defer close(chunkReporter.chunkCh) id, kind := unit.SourceUnitID() ctx := context.WithValues(ctx, "unit_kind", kind, "unit", id) ctx.Logger().V(3).Info("chunking unit") + defer common.RecoverWithHandler(ctx, func(err error) { + report.ReportError(Fatal{ChunkError{Unit: unit, Err: err}}) + catchFirstFatal(Fatal{err}) + }) if err := source.ChunkUnit(ctx, unit, chunkReporter); err != nil { report.ReportError(Fatal{ChunkError{Unit: unit, Err: err}}) catchFirstFatal(Fatal{err}) @@ -597,11 +613,14 @@ func (s *SourceManager) scanWithUnit(ctx context.Context, source SourceUnitChunk var chunkErr error go func() { report.StartUnitChunking(unit, time.Now()) - // TODO: Catch panics and add to report. defer close(chunkReporter.chunkCh) id, kind := unit.SourceUnitID() ctx := context.WithValues(ctx, "unit_kind", kind, "unit", id) ctx.Logger().V(3).Info("chunking unit") + defer common.RecoverWithHandler(ctx, func(err error) { + report.ReportError(Fatal{ChunkError{Unit: unit, Err: err}}) + chunkErr = Fatal{err} + }) if err := source.ChunkUnit(ctx, unit, chunkReporter); err != nil { report.ReportError(Fatal{ChunkError{Unit: unit, Err: err}}) chunkErr = Fatal{err} diff --git a/pkg/sources/source_manager_test.go b/pkg/sources/source_manager_test.go index c2dadd403..3cbfc025d 100644 --- a/pkg/sources/source_manager_test.go +++ b/pkg/sources/source_manager_test.go @@ -83,11 +83,11 @@ func (c *counterChunker) ChunkUnit(ctx context.Context, unit SourceUnit, reporte } // Chunk method that always returns an error. -type errorChunker struct{ error } +type errorChunker struct{ cb func() error } -func (c errorChunker) Chunks(context.Context, chan *Chunk, ...ChunkingTarget) error { return c } -func (c errorChunker) Enumerate(context.Context, UnitReporter) error { return c } -func (c errorChunker) ChunkUnit(context.Context, SourceUnit, ChunkReporter) error { return c } +func (c errorChunker) Chunks(context.Context, chan *Chunk, ...ChunkingTarget) error { return c.cb() } +func (c errorChunker) Enumerate(context.Context, UnitReporter) error { return c.cb() } +func (c errorChunker) ChunkUnit(context.Context, SourceUnit, ChunkReporter) error { return c.cb() } // buildDummy is a helper function to enroll a DummySource with a SourceManager. func buildDummy(chunkMethod chunker) (Source, error) { @@ -147,7 +147,18 @@ func TestSourceManagerWait(t *testing.T) { func TestSourceManagerError(t *testing.T) { mgr := NewManager() - source, err := buildDummy(errorChunker{fmt.Errorf("oops")}) + source, err := buildDummy(errorChunker{func() error { return fmt.Errorf("oops") }}) + assert.NoError(t, err) + ref, err := mgr.EnumerateAndScan(context.Background(), "dummy", source) + assert.NoError(t, err) + <-ref.Done() + assert.Error(t, ref.Snapshot().FatalError()) + assert.Error(t, mgr.Wait()) +} + +func TestSourceManagerPanic(t *testing.T) { + mgr := NewManager() + source, err := buildDummy(errorChunker{func() error { panic("whoops!") }}) assert.NoError(t, err) ref, err := mgr.EnumerateAndScan(context.Background(), "dummy", source) assert.NoError(t, err) @@ -242,6 +253,7 @@ func (c *unitChunker) Chunks(ctx context.Context, ch chan *Chunk, _ ...ChunkingT } return nil } + func (c *unitChunker) Enumerate(ctx context.Context, rep UnitReporter) error { for _, step := range c.steps { if err := rep.UnitOk(ctx, CommonSourceUnit{ID: step.unit}); err != nil { @@ -250,6 +262,7 @@ func (c *unitChunker) Enumerate(ctx context.Context, rep UnitReporter) error { } return nil } + func (c *unitChunker) ChunkUnit(ctx context.Context, unit SourceUnit, rep ChunkReporter) error { for _, step := range c.steps { id, _ := unit.SourceUnitID()