fix error handling

Author: FGasper

internal/verifier/check.go

@@ -3,13 +3,13 @@ package verifier
 import (
 	"context"
 	"fmt"
-	"sync"
 	"time"
 
 	mapset "github.com/deckarep/golang-set/v2"
 	"github.com/pkg/errors"
 	"go.mongodb.org/mongo-driver/bson"
 	"go.mongodb.org/mongo-driver/mongo"
+	"golang.org/x/sync/errgroup"
 )
 
 type GenerationStatus string
@@ -57,19 +57,14 @@ func (verifier *Verifier) waitForChangeStream(ctx context.Context) error {
 	return nil
 }
 
-func (verifier *Verifier) CheckWorker(ctx context.Context) error {
-	verifier.logger.Debug().Msgf("Starting %d verification workers", verifier.numWorkers)
-	ctx, cancel := context.WithCancel(ctx)
-
-	wg := sync.WaitGroup{}
-	for i := 0; i < verifier.numWorkers; i++ {
-		wg.Add(1)
-		go verifier.Work(ctx, i, &wg)
-		time.Sleep(10 * time.Millisecond)
-	}
-
+func (verifier *Verifier) CheckWorker(ctxIn context.Context) error {
 	generation := verifier.generation
 
+	verifier.logger.Debug().
+		Int("generation", generation).
+		Int("workersCount", verifier.numWorkers).
+		Msgf("Starting verification worker threads.")
+
 	// Since we do a progress report right at the start we don’t need
 	// this to go in non-debug output.
 	startLabel := fmt.Sprintf("Starting check generation %d", generation)
@@ -81,44 +76,75 @@ func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 
 	verifier.writeStringBuilder(genStartReport)
 
-	waitForTaskCreation := 0
+	eg, ctx := errgroup.WithContext(ctxIn)
 
-	for {
+	// If the change stream fails, everything should stop.
+	eg.Go(func() error {
 		select {
 		case err := <-verifier.changeStreamErrChan:
-			cancel()
 			return errors.Wrap(err, "change stream failed")
 		case <-ctx.Done():
-			cancel()
 			return nil
-		default:
 		}
+	})
 
-		verificationStatus, err := verifier.GetVerificationStatus()
-		if err != nil {
-			verifier.logger.Error().Msgf("Failed getting verification status: %v", err)
-		}
+	// Start the worker threads.
+	for i := 0; i < verifier.numWorkers; i++ {
+		eg.Go(func() error {
+			return errors.Wrapf(
+				verifier.Work(ctx, i),
+				"worker %d failed",
+				i,
+			)
+		})
+		time.Sleep(10 * time.Millisecond)
+	}
 
-		if waitForTaskCreation%2 == 0 {
-			if generation > 0 || verifier.gen0PendingCollectionTasks.Load() == 0 {
-				verifier.PrintVerificationSummary(ctx, GenerationInProgress)
+	waitForTaskCreation := 0
+
+	eg.Go(func() error {
+		for {
+			verificationStatus, err := verifier.GetVerificationStatus(ctx)
+			if err != nil {
+				return errors.Wrapf(
+					err,
+					"failed to retrieve status of generation %d's tasks",
+					generation,
+				)
 			}
-		}
 
-		//wait for task to be created, if none of the tasks found.
-		if verificationStatus.AddedTasks > 0 || verificationStatus.ProcessingTasks > 0 || verificationStatus.RecheckTasks > 0 {
-			waitForTaskCreation++
-			time.Sleep(verifier.verificationStatusCheckInterval)
-		} else {
-			verifier.PrintVerificationSummary(ctx, GenerationComplete)
-			verifier.logger.Debug().Msg("Verification tasks complete")
-			cancel()
-			wg.Wait()
-			break
+			if waitForTaskCreation%2 == 0 {
+				if generation > 0 || verifier.gen0PendingCollectionTasks.Load() == 0 {
+					verifier.PrintVerificationSummary(ctx, GenerationInProgress)
+				}
+			}
+
+			// The generation continues as long as >=1 task for this generation is
+			// “added” or “pending”.
+			if verificationStatus.AddedTasks > 0 || verificationStatus.ProcessingTasks > 0 {
+				waitForTaskCreation++
+				time.Sleep(verifier.verificationStatusCheckInterval)
+				continue
+			} else {
+				verifier.PrintVerificationSummary(ctx, GenerationComplete)
+				return nil
+			}
 		}
+	})
+
+	err := eg.Wait()
+
+	if err != nil {
+		verifier.logger.Debug().
+			Int("generation", generation).
+			Msgf("Check finished.")
 	}
-	verifier.logger.Debug().Msgf("Check generation %d finished", generation)
-	return nil
+
+	return errors.Wrapf(
+		err,
+		"check generation %d failed",
+		generation,
+	)
 }
 
 func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any, testChan ...chan struct{}) error {
@@ -179,10 +205,14 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 			return errors.Wrap(err, "failed to start change stream on source")
 		}
 	}
-	// Log out the verification status when initially booting up so it's easy to see the current state
-	verificationStatus, err := verifier.GetVerificationStatus()
+
+	// Log the verification status when initially booting up so it's easy to see the current state
+	verificationStatus, err := verifier.GetVerificationStatus(ctx)
 	if err != nil {
-		verifier.logger.Error().Msgf("Failed getting verification status: %v", err)
+		return errors.Wrapf(
+			err,
+			"failed to retrieve verification status",
+		)
 	} else {
 		verifier.logger.Debug().Msgf("Initial verification phase: %+v", verificationStatus)
 	}
@@ -259,8 +289,9 @@ func (verifier *Verifier) CheckDriver(ctx context.Context, filter map[string]any
 
 		err = verifier.ClearRecheckDocs(ctx)
 		if err != nil {
-			verifier.logger.Error().Msgf("Failed trying to clear out old recheck docs, continuing: %v",
-				err)
+			verifier.logger.Warn().
+				Err(err).
+				Msg("Failed to clear out old recheck docs. (This is probably unimportant.)")
 		}
 		verifier.mux.Unlock()
 	}
@@ -368,9 +399,8 @@ func FetchFailedAndIncompleteTasks(ctx context.Context, coll *mongo.Collection,
 	return FailedTasks, IncompleteTasks, nil
 }
 
-func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.WaitGroup) {
-	defer wg.Done()
-
+// Work is the logic for an individual worker thread.
+func (verifier *Verifier) Work(ctx context.Context, workerNum int) error {
 	verifier.logger.Debug().
 		Int("workerNum", workerNum).
 		Msg("Worker started.")
@@ -382,9 +412,9 @@ func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.Wait
 	for {
 		select {
 		case <-ctx.Done():
-			return
+			return ctx.Err()
 		default:
-			task, err := verifier.FindNextVerifyTaskAndUpdate()
+			task, err := verifier.FindNextVerifyTaskAndUpdate(ctx)
 			if errors.Is(err, mongo.ErrNoDocuments) {
 				duration := verifier.workerSleepDelayMillis * time.Millisecond
 
@@ -399,20 +429,32 @@ func (verifier *Verifier) Work(ctx context.Context, workerNum int, wg *sync.Wait
 
 				continue
 			} else if err != nil {
-				panic(err)
+				return errors.Wrap(
+					err,
+					"failed to seek next task",
+				)
 			}
 
-			if task.Type == verificationTaskVerifyCollection {
-				verifier.ProcessCollectionVerificationTask(ctx, workerNum, task)
+			switch task.Type {
+			case verificationTaskVerifyCollection:
+				err := verifier.ProcessCollectionVerificationTask(ctx, workerNum, task)
+				if err != nil {
+					return err
+				}
 				if task.Generation == 0 {
 					newVal := verifier.gen0PendingCollectionTasks.Add(-1)
 					if newVal == 0 {
 						verifier.PrintVerificationSummary(ctx, Gen0MetadataAnalysisComplete)
 					}
 				}
-			} else {
-				verifier.ProcessVerifyTask(workerNum, task)
+			case verificationTaskVerifyDocuments:
+				err := verifier.ProcessVerifyTask(ctx, workerNum, task)
+				if err != nil {
+					return err
+				}
 			}
 		}
 	}
+
+	return nil
 }