add more detail to retries

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()
+					ri.NoteSuccess("received an ID partition")
 				}
 
 				return cursor.Err()

internal/retry/retry.go

@@ -69,11 +69,14 @@ func (r *Retryer) runRetryLoop(
 	li := &LoopInfo{
 		durationLimit: r.retryLimit,
 	}
-	funcinfos := lo.RepeatBy(
-		len(r.callbacks),
-		func(_ int) *FuncInfo {
+	funcinfos := lo.Map(
+		r.callbacks,
+		func(cb retryCallbackInfo, _ int) *FuncInfo {
 			return &FuncInfo{
-				lastResetTime:   msync.NewTypedAtomic(startTime),
+				lastReset: msync.NewTypedAtomic(lastResetInfo{
+					time: startTime,
+				}),
+				description:     cb.description,
 				loopDescription: r.description,
 				loopInfo:        li,
 			}
@@ -113,17 +116,25 @@ func (r *Retryer) runRetryLoop(
 					defer ticker.Stop()
 
 					for {
-						lastSuccessTime := funcinfos[i].lastResetTime.Load()
+						lastReset := funcinfos[i].lastReset.Load()
 
 						select {
 						case <-cbDoneChan:
 							return
 						case <-ticker.C:
-							if funcinfos[i].lastResetTime.Load() == lastSuccessTime {
-								logger.Warn().
+							if funcinfos[i].lastReset.Load() == lastReset {
+								event := logger.Warn().
 									Str("callbackDescription", curCbInfo.description).
-									Time("lastSuccessAt", lastSuccessTime).
-									Str("elapsedTime", reportutils.DurationToHMS(time.Since(lastSuccessTime))).
+									Time("since", lastReset.time)
+
+								if successDesc, hasDesc := lastReset.description.Get(); hasDesc {
+									event.
+										Str("successDescription", successDesc).
+										Uint64("successesSoFar", lastReset.resetsSoFar)
+								}
+
+								event.
+									Str("elapsedTime", reportutils.DurationToHMS(time.Since(lastReset.time))).
 									Msg("Operation has not reported success for a while.")
 							}
 						}
@@ -164,9 +175,11 @@ 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, r.callbacks[groupErr.funcNum], groupErr.errFromCallback) {
+		if !r.shouldRetryWithSleep(logger, sleepTime, descriptions, cbErr) {
 			return groupErr.errFromCallback
 		}
 
@@ -201,7 +214,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.lastResetTime.Store(now)
+				curInfo.lastReset.Store(lastResetInfo{time: now})
 			}
 		}
 	}
@@ -235,7 +248,7 @@ func (r *Retryer) addDescriptionToEvent(event *zerolog.Event) *zerolog.Event {
 func (r *Retryer) shouldRetryWithSleep(
 	logger *logger.Logger,
 	sleepTime time.Duration,
-	cbInfo retryCallbackInfo,
+	descriptions []string,
 	err error,
 ) bool {
 	if err == nil {
@@ -266,11 +279,7 @@ func (r *Retryer) shouldRetryWithSleep(
 		),
 	)
 
-	if loopDesc, hasLoopDesc := r.description.Get(); hasLoopDesc {
-		event.Str("operationDescription", loopDesc)
-	}
-
-	event.Str("callbackDescription", cbInfo.description).
+	event.Strs("description", descriptions).
 		Int("error code", util.GetErrorCode(err)).
 		Err(err)
 

internal/retry/retry_info.go

@@ -1,9 +1,11 @@
 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"
@@ -19,11 +21,20 @@ 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]
-	lastResetTime   *msync.TypedAtomic[time.Time]
+
+	lastReset *msync.TypedAtomic[lastResetInfo]
 }
 
 // Log will log a debug-level message for the current Info values and the provided strings.
@@ -69,7 +80,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.lastResetTime.Load())
+	return time.Since(fi.lastReset.Load().time)
 }
 
 // NoteSuccess is used to tell the retry util to reset its measurement
@@ -78,6 +89,21 @@ 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() {
-	i.lastResetTime.Store(time.Now())
+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
 }

internal/retry/retryer_test.go

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

internal/verifier/change_stream.go

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

internal/verifier/change_stream_test.go

@@ -172,7 +172,6 @@ func (suite *IntegrationTestSuite) TestChangeStreamFilter_WithNamespaces() {
 			events = append(events, newEvent)
 		}
 
-		suite.T().Logf("Change stream op time (got event? %v): %v", gotEvent, csOpTime)
 		if csOpTime.After(*changeStreamStopTime) {
 			break
 		}

internal/verifier/compare.go

@@ -324,7 +324,7 @@ func (verifier *Verifier) getFetcherChannelsAndCallbacks(
 		)
 
 		if err == nil {
-			state.NoteSuccess()
+			state.NoteSuccess("opened src find cursor")
 
 			err = errors.Wrap(
 				iterateCursorToChannel(ctx, state, cursor, srcChannel),
@@ -350,7 +350,7 @@ func (verifier *Verifier) getFetcherChannelsAndCallbacks(
 		)
 
 		if err == nil {
-			state.NoteSuccess()
+			state.NoteSuccess("opened dst find cursor")
 
 			err = errors.Wrap(
 				iterateCursorToChannel(ctx, state, cursor, dstChannel),
@@ -376,7 +376,7 @@ func iterateCursorToChannel(
 	writer chan<- bson.Raw,
 ) error {
 	for cursor.Next(ctx) {
-		state.NoteSuccess()
+		state.NoteSuccess("received a document")
 		writer <- slices.Clone(cursor.Current)
 	}