Skip to content

REP-5329 Fix retry on document reads. #69

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
FGasper merged 4 commits into from
Dec 4, 2024
Merged

REP-5329 Fix retry on document reads. #69

Show file tree
Hide file tree
Changes from 3 commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
7bb7dfe
fix retry on doc-read failure
FGasper Dec 3, 2024
006d3f9
pointers
FGasper Dec 3, 2024
e21f98f
panics & fix
FGasper Dec 3, 2024
201ffe8
move panics out
FGasper Dec 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 3 additions & 3 deletions internal/partitions/partitions.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -119,7 +119,7 @@ const (
func PartitionCollectionWithSize(
ctx context.Context,
uuidEntry *uuidutil.NamespaceAndUUID,
retryer retry.Retryer,
retryer *retry.Retryer,
srcClient *mongo.Client,
replicatorList []Replicator,
subLogger *logger.Logger,
Expand All @@ -137,7 +137,7 @@ func PartitionCollectionWithSize(
partitions, docCount, byteCount, err := PartitionCollectionWithParameters(
ctx,
uuidEntry,
&retryer,
retryer,
srcClient,
replicatorList,
defaultSampleRate,
Expand All @@ -153,7 +153,7 @@ func PartitionCollectionWithSize(
return PartitionCollectionWithParameters(
ctx,
uuidEntry,
&retryer,
retryer,
srcClient,
replicatorList,
defaultSampleRate,
Expand Down
17 changes: 15 additions & 2 deletions internal/retry/retry.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,9 +44,9 @@ type RetryCallback = func(context.Context, *FuncInfo) error
// This returns an error if the duration limit is reached, or if f() returns a
// non-transient error.
func (r *Retryer) Run(
ctx context.Context, logger *logger.Logger, f ...RetryCallback,
ctx context.Context, logger *logger.Logger, funcs ...RetryCallback,
) error {
return r.runRetryLoop(ctx, logger, f)
return r.runRetryLoop(ctx, logger, funcs)
}

// runRetryLoop contains the core logic for the retry loops.
Expand Down Expand Up @@ -74,10 +74,23 @@ func (r *Retryer) runRetryLoop(
sleepTime := minSleepTime

for {
if beforeFunc, hasBefore := r.before.Get(); hasBefore {
beforeFunc()
}

eg, egCtx := errgroup.WithContext(ctx)
for i, curFunc := range funcs {
if curFunc == nil {
panic("curFunc should be non-nil")
}

eg.Go(func() error {
if curFunc == nil {
Copy link
Collaborator

@tdq45gj tdq45gj Dec 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we check curFunc to be non-nil both twice?

Copy link
Collaborator Author

@FGasper FGasper Dec 4, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This 2nd check is probably redundant. I was just trying to suss out why I was seeing segfaults, and I figured I might as well leave the panics in.

panic("curFunc should be non-nil")
}
if funcinfos[i] == nil {
panic(fmt.Sprintf("funcinfos[%d] should be non-nil", i))
}
err := curFunc(egCtx, funcinfos[i])

if err != nil {
Expand Down
22 changes: 16 additions & 6 deletions internal/retry/retryer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,24 +2,27 @@ package retry

import (
"time"

"github.com/10gen/migration-verifier/option"
)

// Retryer handles retrying operations that fail because of network failures.
type Retryer struct {
retryLimit time.Duration
retryRandomly bool
before option.Option[func()]
additionalErrorCodes []int
}

// New returns a new retryer.
func New(retryLimit time.Duration) Retryer {
func New(retryLimit time.Duration) *Retryer {
return NewWithRandomlyRetries(retryLimit, false)
}

// NewWithRandomlyRetries returns a new retryer, but allows the option of setting the
// retryRandomly field.
func NewWithRandomlyRetries(retryLimit time.Duration, retryRandomly bool) Retryer {
return Retryer{
func NewWithRandomlyRetries(retryLimit time.Duration, retryRandomly bool) *Retryer {
return &Retryer{
retryLimit: retryLimit,
retryRandomly: retryRandomly,
}
Expand All @@ -29,9 +32,16 @@ func NewWithRandomlyRetries(retryLimit time.Duration, retryRandomly bool) Retrye
// this method. This allows for a single function to customize the codes it
// wants to retry on. Note that if the Retryer already has additional custom
// error codes set, these are _replaced_ when this method is called.
func (r Retryer) WithErrorCodes(codes ...int) Retryer {
r2 := r
func (r *Retryer) WithErrorCodes(codes ...int) *Retryer {
r2 := *r
r2.additionalErrorCodes = codes

return r2
return &r2
}

func (r *Retryer) WithBefore(todo func()) *Retryer {
r2 := *r
r2.before = option.Some(todo)

return &r2
}
4 changes: 2 additions & 2 deletions internal/uuidutil/get_uuid.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ type NamespaceAndUUID struct {
CollName string
}

func GetCollectionNamespaceAndUUID(ctx context.Context, logger *logger.Logger, retryer retry.Retryer, db *mongo.Database, collName string) (*NamespaceAndUUID, error) {
func GetCollectionNamespaceAndUUID(ctx context.Context, logger *logger.Logger, retryer *retry.Retryer, db *mongo.Database, collName string) (*NamespaceAndUUID, error) {
binaryUUID, uuidErr := GetCollectionUUID(ctx, logger, retryer, db, collName)
if uuidErr != nil {
return nil, uuidErr
Expand All @@ -39,7 +39,7 @@ func GetCollectionNamespaceAndUUID(ctx context.Context, logger *logger.Logger, r
}, nil
}

func GetCollectionUUID(ctx context.Context, logger *logger.Logger, retryer retry.Retryer, db *mongo.Database, collName string) (*primitive.Binary, error) {
func GetCollectionUUID(ctx context.Context, logger *logger.Logger, retryer *retry.Retryer, db *mongo.Database, collName string) (*primitive.Binary, error) {
filter := bson.D{{"name", collName}}
opts := options.ListCollections().SetNameOnly(false)

Expand Down
43 changes: 26 additions & 17 deletions internal/verifier/compare.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,31 +25,40 @@ func (verifier *Verifier) FetchAndCompareDocuments(
types.ByteCount,
error,
) {
srcChannel, dstChannel, readSrcCallback, readDstCallback := verifier.getFetcherChannelsAndCallbacks(task)
var srcChannel, dstChannel <-chan bson.Raw
var readSrcCallback, readDstCallback func(context.Context, *retry.FuncInfo) error

results := []VerificationResult{}
var docCount types.DocumentCount
var byteCount types.ByteCount

retryer := retry.New(retry.DefaultDurationLimit)

err := retryer.Run(
givenCtx,
verifier.logger,
readSrcCallback,
readDstCallback,
func(ctx context.Context, _ *retry.FuncInfo) error {
var err error
results, docCount, byteCount, err = verifier.compareDocsFromChannels(
ctx,
task,
srcChannel,
dstChannel,
)
err := retryer.
WithBefore(func() {
srcChannel, dstChannel, readSrcCallback, readDstCallback = verifier.getFetcherChannelsAndCallbacks(task)
}).
Comment on lines +38 to +40
Copy link
Collaborator

@tdq45gj tdq45gj Dec 3, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm trying to understand why do we need before func. Is it the same as calling the function at the start of the retried function except that the error from it won't be retried?

Copy link
Collaborator Author

@FGasper FGasper Dec 4, 2024

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The trick is that there are multiple retried funcs. Having a “before” func guarantees execution of one piece before all the others … which is crucial here because the callbacks & channels have to get recreated every time.

I’ll clarify this in docs.

Run(
givenCtx,
verifier.logger,
func(ctx context.Context, fi *retry.FuncInfo) error {
return readSrcCallback(ctx, fi)
},
func(ctx context.Context, fi *retry.FuncInfo) error {
return readDstCallback(ctx, fi)
},
func(ctx context.Context, _ *retry.FuncInfo) error {
var err error
results, docCount, byteCount, err = verifier.compareDocsFromChannels(
ctx,
task,
srcChannel,
dstChannel,
)

return err
},
)
return err
},
)

return results, docCount, byteCount, err
}
Expand Down
4 changes: 2 additions & 2 deletions internal/verifier/mongos_refresh.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -137,7 +137,7 @@ func RefreshAllMongosInstances(
func getAnyExistingShardConnectionStr(
ctx context.Context,
l *logger.Logger,
r retry.Retryer,
r *retry.Retryer,
client *mongo.Client,
) (string, error) {
res, err := runListShards(ctx, l, r, client)
Expand Down Expand Up @@ -169,7 +169,7 @@ func getAnyExistingShardConnectionStr(
func runListShards(
ctx context.Context,
l *logger.Logger,
r retry.Retryer,
r *retry.Retryer,
client *mongo.Client,
) (*mongo.SingleResult, error) {
var res *mongo.SingleResult
Expand Down
Loading