revert retryer changes

Author: FGasper

internal/partitions/partitions.go

@@ -621,7 +621,7 @@ func getMidIDBounds(
 
 					// Append the copied bound to the other mid _id bounds.
 					midIDBounds = append(midIDBounds, bound)
-					ri.NoteSuccess("received an ID partition")
+					ri.NoteSuccess()
 				}
 
 				return cursor.Err()

internal/retry/retry.go

@@ -69,14 +69,11 @@ func (r *Retryer) runRetryLoop(
 	li := &LoopInfo{
 		durationLimit: r.retryLimit,
 	}
-	funcinfos := lo.Map(
-		r.callbacks,
-		func(cb retryCallbackInfo, _ int) *FuncInfo {
+	funcinfos := lo.RepeatBy(
+		len(r.callbacks),
+		func(_ int) *FuncInfo {
 			return &FuncInfo{
-				lastReset: msync.NewTypedAtomic(lastResetInfo{
-					time: startTime,
-				}),
-				description:     cb.description,
+				lastResetTime:   msync.NewTypedAtomic(startTime),
 				loopDescription: r.description,
 				loopInfo:        li,
 			}
@@ -116,25 +113,17 @@ func (r *Retryer) runRetryLoop(
 					defer ticker.Stop()
 
 					for {
-						lastReset := funcinfos[i].lastReset.Load()
+						lastSuccessTime := funcinfos[i].lastResetTime.Load()
 
 						select {
 						case <-cbDoneChan:
 							return
 						case <-ticker.C:
-							if funcinfos[i].lastReset.Load() == lastReset {
-								event := logger.Warn().
+							if funcinfos[i].lastResetTime.Load() == lastSuccessTime {
+								logger.Warn().
 									Str("callbackDescription", curCbInfo.description).
-									Time("noSuccessSince", lastReset.time).
-									Uint64("successesSoFar", lastReset.resetsSoFar)
-
-								if successDesc, hasDesc := lastReset.description.Get(); hasDesc {
-									event.
-										Str("lastSuccessDescription", successDesc)
-								}
-
-								event.
-									Str("elapsedTime", reportutils.DurationToHMS(time.Since(lastReset.time))).
+									Time("lastSuccessAt", lastSuccessTime).
+									Str("elapsedTime", reportutils.DurationToHMS(time.Since(lastSuccessTime))).
 									Msg("Operation has not reported success for a while.")
 							}
 						}
@@ -175,11 +164,9 @@ func (r *Retryer) runRetryLoop(
 		}
 
 		failedFuncInfo := funcinfos[groupErr.funcNum]
-		descriptions := failedFuncInfo.GetDescriptions()
-		cbErr := groupErr.errFromCallback
 
 		// Not a transient error? Fail immediately.
-		if !r.shouldRetryWithSleep(logger, sleepTime, descriptions, cbErr) {
+		if !r.shouldRetryWithSleep(logger, sleepTime, *failedFuncInfo, groupErr.errFromCallback) {
 			return groupErr.errFromCallback
 		}
 
@@ -214,7 +201,7 @@ func (r *Retryer) runRetryLoop(
 		// Set all of the funcs that did *not* fail as having just succeeded.
 		for i, curInfo := range funcinfos {
 			if i != groupErr.funcNum {
-				curInfo.lastReset.Store(lastResetInfo{time: now})
+				curInfo.lastResetTime.Store(now)
 			}
 		}
 	}
@@ -248,7 +235,7 @@ func (r *Retryer) addDescriptionToEvent(event *zerolog.Event) *zerolog.Event {
 func (r *Retryer) shouldRetryWithSleep(
 	logger *logger.Logger,
 	sleepTime time.Duration,
-	descriptions []string,
+	funcinfo FuncInfo,
 	err error,
 ) bool {
 	if err == nil {
@@ -263,35 +250,26 @@ func (r *Retryer) shouldRetryWithSleep(
 	)
 
 	event := logger.WithLevel(
-		lo.Ternary(
-			// If it’s transient, surface it as info.
-			isTransient,
-			zerolog.InfoLevel,
-
-			lo.Ternary(
-				// Context cancellation is unimportant, so debug.
-				errors.Is(err, context.Canceled),
-				zerolog.DebugLevel,
-
-				// Other non-retryables are serious, so warn.
-				zerolog.WarnLevel,
-			),
-		),
+		lo.Ternary(isTransient, zerolog.InfoLevel, zerolog.WarnLevel),
 	)
 
-	event.Strs("description", descriptions).
+	if loopDesc, hasLoopDesc := r.description.Get(); hasLoopDesc {
+		event.Str("operationDescription", loopDesc)
+	}
+
+	event.Str("callbackDescription", funcinfo.description).
 		Int("error code", util.GetErrorCode(err)).
 		Err(err)
 
 	if isTransient {
 		event.
 			Stringer("delay", sleepTime).
-			Msg("Got retryable error. Pausing, then will retry.")
+			Msg("Pausing before retrying after transient error.")
 
 		return true
 	}
 
-	event.Msg("Non-retryable error occurred.")
+	event.Msg("Non-transient error occurred.")
 
 	return false
 }

internal/retry/retry_info.go

@@ -1,11 +1,9 @@
 package retry
 
 import (
-	"slices"
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/reportutils"
-	"github.com/10gen/migration-verifier/mslices"
 	"github.com/10gen/migration-verifier/msync"
 	"github.com/10gen/migration-verifier/option"
 	"github.com/rs/zerolog"
@@ -21,20 +19,11 @@ type LoopInfo struct {
 	durationLimit time.Duration
 }
 
-type lastResetInfo struct {
-	time time.Time
-
-	// These go into logs to facilitate debugging.
-	description option.Option[string]
-	resetsSoFar uint64
-}
-
 type FuncInfo struct {
 	loopInfo        *LoopInfo
 	description     string
 	loopDescription option.Option[string]
-
-	lastReset *msync.TypedAtomic[lastResetInfo]
+	lastResetTime   *msync.TypedAtomic[time.Time]
 }
 
 // Log will log a debug-level message for the current Info values and the provided strings.
@@ -80,7 +69,7 @@ func (fi *FuncInfo) GetAttemptNumber() int {
 // GetDurationSoFar returns the Info's current duration so far. This duration
 // applies to the duration of retrying for transient errors only.
 func (fi *FuncInfo) GetDurationSoFar() time.Duration {
-	return time.Since(fi.lastReset.Load().time)
+	return time.Since(fi.lastResetTime.Load())
 }
 
 // NoteSuccess is used to tell the retry util to reset its measurement
@@ -89,21 +78,6 @@ func (fi *FuncInfo) GetDurationSoFar() time.Duration {
 //
 // Call this after every successful command in a multi-command callback.
 // (It’s useless--but harmless--in a single-command callback.)
-func (i *FuncInfo) NoteSuccess(description string) {
-	totalResets := i.lastReset.Load().resetsSoFar
-
-	i.lastReset.Store(lastResetInfo{
-		description: option.Some(description),
-		time:        time.Now(),
-		resetsSoFar: 1 + totalResets,
-	})
-}
-
-func (i *FuncInfo) GetDescriptions() []string {
-	descriptions := mslices.Of(i.description)
-	if loopDesc, hasDesc := i.loopDescription.Get(); hasDesc {
-		descriptions = slices.Insert(descriptions, 0, loopDesc)
-	}
-
-	return descriptions
+func (i *FuncInfo) NoteSuccess() {
+	i.lastResetTime.Store(time.Now())
 }

internal/retry/retryer_test.go

@@ -8,7 +8,6 @@ import (
 	"time"
 
 	"github.com/10gen/migration-verifier/internal/util"
-	"github.com/10gen/migration-verifier/option"
 	"go.mongodb.org/mongo-driver/mongo"
 )
 
@@ -100,11 +99,8 @@ func (suite *UnitTestSuite) TestRetryerDurationReset() {
 	noSuccessIterations := 0
 	f1 := func(_ context.Context, ri *FuncInfo) error {
 		// Artificially advance how much time was taken.
-		ri.lastReset.Store(
-			lastResetInfo{
-				time:        ri.lastReset.Load().time.Add(-2 * ri.loopInfo.durationLimit),
-				description: option.Some("artificially rewinding time"),
-			},
+		ri.lastResetTime.Store(
+			ri.lastResetTime.Load().Add(-2 * ri.loopInfo.durationLimit),
 		)
 
 		noSuccessIterations++
@@ -128,13 +124,12 @@ func (suite *UnitTestSuite) TestRetryerDurationReset() {
 	successIterations := 0
 	f2 := func(_ context.Context, ri *FuncInfo) error {
 		// Artificially advance how much time was taken.
-		ri.lastReset.Store(
-			lastResetInfo{
-				time:        ri.lastReset.Load().time.Add(-2 * ri.loopInfo.durationLimit),
-				description: option.Some("artificially rewinding time"),
-			},
+		ri.lastResetTime.Store(
+			ri.lastResetTime.Load().Add(-2 * ri.loopInfo.durationLimit),
 		)
 
+		ri.NoteSuccess()
+
 		successIterations++
 		if successIterations == 1 {
 			return someNetworkError
@@ -312,7 +307,7 @@ func (suite *UnitTestSuite) TestMulti_LongRunningSuccess() {
 
 	err := retryer.WithCallback(
 		func(ctx context.Context, fi *FuncInfo) error {
-			fi.NoteSuccess("success right away")
+			fi.NoteSuccess()
 
 			if time.Now().Before(succeedPastTime) {
 				time.Sleep(1 * time.Second)

internal/verifier/change_stream.go

@@ -298,7 +298,7 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 		eventsRead++
 	}
 
-	ri.NoteSuccess("received a batch of change events")
+	ri.NoteSuccess()
 
 	if eventsRead == 0 {
 		return nil

internal/verifier/compare.go

@@ -324,7 +324,7 @@ func (verifier *Verifier) getFetcherChannelsAndCallbacks(
 		)
 
 		if err == nil {
-			state.NoteSuccess("opened src find cursor")
+			state.NoteSuccess()
 
 			err = errors.Wrap(
 				iterateCursorToChannel(ctx, state, cursor, srcChannel),
@@ -350,7 +350,7 @@ func (verifier *Verifier) getFetcherChannelsAndCallbacks(
 		)
 
 		if err == nil {
-			state.NoteSuccess("opened dst find cursor")
+			state.NoteSuccess()
 
 			err = errors.Wrap(
 				iterateCursorToChannel(ctx, state, cursor, dstChannel),
@@ -378,7 +378,7 @@ func iterateCursorToChannel(
 	defer close(writer)
 
 	for cursor.Next(ctx) {
-		state.NoteSuccess("received a document")
+		state.NoteSuccess()
 
 		select {
 		case <-ctx.Done():