Read docs in parallel

Author: FGasper

internal/verifier/compare.go

@@ -146,14 +146,15 @@ func (verifier *Verifier) compareDocsFromChannels(
 	// lags the other significantly, we won’t keep caching the faster side’s
 	// documents and thus consume more & more memory.
 	for !srcClosed || !dstClosed {
-		if !srcClosed {
-			simpleTimerReset(readTimer, readTimeout)
+		simpleTimerReset(readTimer, readTimeout)
 
+		eg, egCtx := errgroup.WithContext(ctx)
+		eg.Go(func() error {
 			select {
-			case <-ctx.Done():
-				return nil, 0, 0, ctx.Err()
+			case <-egCtx.Done():
+				return egCtx.Err()
 			case <-readTimer.C:
-				return nil, 0, 0, errors.Errorf(
+				return errors.Errorf(
 					"failed to read from source after %s",
 					readTimeout,
 				)
@@ -168,42 +169,53 @@ func (verifier *Verifier) compareDocsFromChannels(
 				err := handleNewDoc(doc, true)
 
 				if err != nil {
-					return nil, 0, 0, errors.Wrapf(
+					return errors.Wrapf(
 						err,
 						"comparer thread failed to handle source doc with ID %v",
 						doc.Lookup("_id"),
 					)
 				}
 			}
-		}
 
-		if !dstClosed {
-			simpleTimerReset(readTimer, readTimeout)
+			return nil
+		})
 
-			select {
-			case <-ctx.Done():
-				return nil, 0, 0, ctx.Err()
-			case <-readTimer.C:
-				return nil, 0, 0, errors.Errorf(
-					"failed to read from destination after %s",
-					readTimeout,
-				)
-			case doc, alive := <-dstChannel:
-				if !alive {
-					dstClosed = true
-					break
+		if !dstClosed {
+			eg.Go(func() error {
+				select {
+				case <-egCtx.Done():
+					return egCtx.Err()
+				case <-readTimer.C:
+					return errors.Errorf(
+						"failed to read from destination after %s",
+						readTimeout,
+					)
+				case doc, alive := <-dstChannel:
+					if !alive {
+						dstClosed = true
+						break
+					}
+
+					err := handleNewDoc(doc, false)
+
+					if err != nil {
+						return errors.Wrapf(
+							err,
+							"comparer thread failed to handle destination doc with ID %v",
+							doc.Lookup("_id"),
+						)
+					}
 				}
 
-				err := handleNewDoc(doc, false)
+				return nil
+			})
+		}
 
-				if err != nil {
-					return nil, 0, 0, errors.Wrapf(
-						err,
-						"comparer thread failed to handle destination doc with ID %v",
-						doc.Lookup("_id"),
-					)
-				}
-			}
+		if err := eg.Wait(); err != nil {
+			return nil, 0, 0, errors.Wrap(
+				err,
+				"failed to compare documents",
+			)
 		}
 	}