rename

internal/util/ctxutil.go

@@ -7,22 +7,34 @@ import (
 	"github.com/pkg/errors"
 )
 
-// Returns an error that combines a context’s error with
-// the error’s cause. If those are the same thing, then
-// just the error is returned.
+// Err returns the error from the given context.
 //
-// The resulting error will be something like:
+// Unlike in the standard library, this Err() is a wrapper around:
+// - the underlying context’s Err(), and
+// - the "cause" from a cancellation, timeout, or deadline
 //
-//	context canceled: got unrecognized change event type
+// Thus, errors.Is() will still work against built-in errors
+// (e.g., context.Canceled).
 //
-// NB: The returned error wraps both the context’s original error
-// *and* the error’s cause.
-func WrapCtxErrWithCause(ctx context.Context) error {
+// NB: This is copied from mongosync.
+func Err(ctx context.Context) error {
 	cause := context.Cause(ctx)
+	err := ctx.Err() //nolint:gocritic
 
-	if errors.Is(cause, ctx.Err()) {
-		return ctx.Err()
+	if cause == nil {
+		return err
 	}
 
-	return fmt.Errorf("%w: %w", ctx.Err(), cause)
+	// Usually the cause is distinct from the stdlib’s error. Sometimes, though,
+	// you might have something like
+	// `cancel(fmt.Errorf("All done (%w)", context.Canceled))`. In this case
+	// the cause wraps the stdlib error, so there’s no point in wrapping again.
+	// (It reads oddly: `context canceled: All done (context canceled)`.)
+	//
+	// Thus, we just return the cause here.
+	if errors.Is(cause, err) {
+		return cause
+	}
+
+	return fmt.Errorf("%w: %w", err, cause)
 }

internal/verifier/change_stream.go

@@ -129,7 +129,7 @@ func (verifier *Verifier) RunChangeEventHandler(ctx context.Context, reader *Cha
 	for err == nil {
 		select {
 		case <-ctx.Done():
-			err = util.WrapCtxErrWithCause(ctx)
+			err = util.Err(ctx)
 
 			verifier.logger.Debug().
 				Err(err).
@@ -377,7 +377,7 @@ func (csr *ChangeStreamReader) readAndHandleOneChangeEventBatch(
 
 	select {
 	case <-ctx.Done():
-		return util.WrapCtxErrWithCause(ctx)
+		return util.Err(ctx)
 	case <-csr.handlerError.Ready():
 		return csr.wrapHandlerErrorForReader()
 	case csr.changeEventBatchChan <- changeEventBatch:
@@ -424,7 +424,7 @@ func (csr *ChangeStreamReader) iterateChangeStream(
 
 		// If the context is canceled, return immmediately.
 		case <-ctx.Done():
-			err := util.WrapCtxErrWithCause(ctx)
+			err := util.Err(ctx)
 
 			csr.logger.Debug().
 				Err(err).

internal/verifier/check.go

@@ -50,7 +50,7 @@ func (verifier *Verifier) Check(ctx context.Context, filter map[string]any) {
 func (verifier *Verifier) waitForChangeStream(ctx context.Context, csr *ChangeStreamReader) error {
 	select {
 	case <-ctx.Done():
-		return util.WrapCtxErrWithCause(ctx)
+		return util.Err(ctx)
 	case <-csr.readerError.Ready():
 		err := csr.readerError.Get()
 		verifier.logger.Warn().Err(err).