[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

@@ -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 {

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}

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()