fixes

Author: FGasper

internal/documentmap/documentmap.go

@@ -126,9 +126,9 @@ func (m *Map) ImportFromCursor(ctx context.Context, cursor *mongo.Cursor, tracke
 
 	m.bytesSize = bytesReturned
 
-	m.logger.Debug().
+	m.logger.Info().
 		Int("documentedReturned", nDocumentsReturned).
-		Str("totalSize", reportutils.BytesToUnit(bytesReturned, reportutils.FindBestUnit(bytesReturned))).
+		Str("totalSize", reportutils.FmtBytes(bytesReturned)).
 		Msgf("Finished reading %#q query.", "find")
 
 	return nil

internal/memorytracker/memorytracker.go

@@ -23,7 +23,10 @@ type Tracker struct {
 }
 
 func Start(ctx context.Context, logger *logger.Logger, max Unit) *Tracker {
-	tracker := Tracker{max: max}
+	tracker := Tracker{
+		max:    max,
+		logger: logger,
+	}
 
 	go tracker.track(ctx)
 
@@ -38,7 +41,7 @@ func (mt *Tracker) AddWriter() Writer {
 
 	mt.selectCases = append(mt.selectCases, reflect.SelectCase{
 		Dir:  reflect.SelectRecv,
-		Chan: reflect.ValueOf(newChan),
+		Chan: reflect.ValueOf(reader(newChan)),
 	})
 
 	return newChan
@@ -70,7 +73,7 @@ func (mt *Tracker) removeSelectCase(i int) {
 
 func (mt *Tracker) track(ctx context.Context) {
 	for {
-		if mt.cur <= mt.max {
+		if mt.cur > mt.max {
 			mt.logger.Panic().
 				Int64("usage", mt.cur).
 				Int64("softLimit", mt.max).
@@ -98,11 +101,7 @@ func (mt *Tracker) track(ctx context.Context) {
 				Msg("Reclaimed tracked memory.")
 		}
 
-		if alive {
-			if got == 0 {
-				mt.logger.Panic().Msg("Got zero track value but channel is not closed.")
-			}
-		} else {
+		if !alive {
 			if got != 0 {
 				mt.logger.Panic().
 					Int64("receivedValue", got).

internal/reportutils/reportutils.go

@@ -70,7 +70,8 @@ func DurationToHMS(duration time.Duration) string {
 // FmtBytes is a convenience that combines BytesToUnit with FindBestUnit.
 // Use it to format a single count of bytes.
 func FmtBytes[T num16Plus](count T) string {
-	return BytesToUnit(count, FindBestUnit(count))
+	unit := FindBestUnit(count)
+	return BytesToUnit(count, unit) + " " + string(unit)
 }
 
 // BytesToUnit returns a stringified number that represents `count`

internal/verifier/check.go

@@ -63,7 +63,7 @@ func (verifier *Verifier) waitForChangeStream() error {
 
 func (verifier *Verifier) CheckWorker(ctx context.Context) error {
 	verifier.logger.Debug().Msgf("Starting %d verification workers", verifier.numWorkers)
-	memTracker := memorytracker.Start(ctx, verifier.logger, 40_000_000) // TODO
+	memTracker := memorytracker.Start(ctx, verifier.logger, 40_000_000_000) // TODO
 	ctx, cancel := context.WithCancel(ctx)
 	wg := sync.WaitGroup{}
 	for i := 0; i < verifier.numWorkers; i++ {

internal/verifier/migration_verifier.go

@@ -504,8 +504,8 @@ func (verifier *Verifier) fetchDocuments(task *VerificationTask, trackerWriter m
 
 		if err == nil && int64(srcClientMap.TotalDocsBytes()) > warnThreshold {
 			verifier.logger.Warn().
-				Str("totalSize", reportutils.BytesToUnit(srcClientMap.TotalDocsBytes(), reportutils.FindBestUnit(srcClientMap.TotalDocsBytes()))).
-				Str("intendedPartitionSize", reportutils.BytesToUnit(verifier.partitionSizeInBytes, reportutils.FindBestUnit(verifier.partitionSizeInBytes))).
+				Str("totalSize", reportutils.FmtBytes(srcClientMap.TotalDocsBytes())).
+				Str("intendedPartitionSize", reportutils.FmtBytes(verifier.partitionSizeInBytes)).
 				Str("filter", fmt.Sprintf("%v", findCmd)).
 				Msg("Partition greatly exceeds desired size. This may cause excess memory usage.")
 		}