release-23.1: kv: implement errors.Wrapper on sendError, deflake test

pkg/kv/kvclient/kvcoord/dist_sender.go

@@ -1766,7 +1766,7 @@ func (ds *DistSender) sendPartialBatch(
 			// deduceRetryEarlyExitError() call below the loop is inhibited.
 			pErr = kvpb.NewError(err)
 			switch {
-			case errors.HasType(err, sendError{}):
+			case IsSendError(err):
 				// We've tried all the replicas without success. Either they're all
 				// down, or we're using an out-of-date range descriptor. Evict from the
 				// cache and try again with an updated descriptor. Re-sending the
@@ -2021,7 +2021,7 @@ func noMoreReplicasErr(ambiguousErr, lastAttemptErr error) error {
 	// one to return; we may want to remember the "best" error we've seen (for
 	// example, a NotLeaseHolderError conveys more information than a
 	// RangeNotFound).
-	return newSendError(fmt.Sprintf("sending to all replicas failed; last error: %s", lastAttemptErr))
+	return newSendError(errors.Wrap(lastAttemptErr, "sending to all replicas failed; last error"))
 }
 
 // defaultSendClosedTimestampPolicy is used when the closed timestamp policy
@@ -2424,7 +2424,7 @@ func (ds *DistSender) sendToReplicas(
 								log.VEventf(
 									ctx, 2, "transport incompatible with updated routing; bailing early",
 								)
-								return nil, newSendError(fmt.Sprintf("leaseholder not found in transport; last error: %s", tErr.Error()))
+								return nil, newSendError(errors.Wrap(tErr, "leaseholder not found in transport; last error"))
 							}
 						}
 					}
@@ -2640,28 +2640,32 @@ func skipStaleReplicas(
 // TODO(andrei): clean up this stuff and tighten the meaning of the different
 // errors.
 type sendError struct {
-	message string
+	cause error
 }
 
-// newSendError creates a sendError.
-func newSendError(msg string) error {
-	return sendError{message: msg}
+// newSendError creates a sendError that wraps the given error.
+func newSendError(err error) error {
+	return &sendError{cause: err}
 }
 
 // TestNewSendError creates a new sendError for the purpose of unit tests
 func TestNewSendError(msg string) error {
-	return newSendError(msg)
+	return newSendError(errors.NewWithDepthf(1, "%s", msg))
 }
 
-// SendErrorString is the prefix for all sendErrors, exported in order to
-// perform cross-node error-checks.
-const SendErrorString = "failed to send RPC"
-
-func (s sendError) Error() string {
-	return SendErrorString + ": " + s.message
+// Error implements error.
+func (s *sendError) Error() string {
+	return fmt.Sprintf("failed to send RPC: %s", s.cause)
 }
 
+// Cause implements errors.Causer.
+// NB: this is an obsolete method, use Unwrap() instead.
+func (s *sendError) Cause() error { return s.cause }
+
+// Unwrap implements errors.Wrapper.
+func (s *sendError) Unwrap() error { return s.cause }
+
 // IsSendError returns true if err is a sendError.
 func IsSendError(err error) bool {
-	return errors.HasType(err, sendError{})
+	return errors.HasType(err, &sendError{})
 }

pkg/kv/kvclient/kvcoord/dist_sender_rangefeed.go

@@ -761,7 +761,7 @@ func (ds *DistSender) singleRangeFeed(
 	for {
 		stuckWatcher.stop() // if timer is running from previous iteration, stop it now
 		if transport.IsExhausted() {
-			return args.Timestamp, newSendError("sending to all replicas failed")
+			return args.Timestamp, newSendError(errors.New("sending to all replicas failed"))
 		}
 		maybeCleanupStream()
 

pkg/kv/kvclient/kvcoord/dist_sender_test.go

@@ -3660,7 +3660,7 @@ func TestMultipleErrorsMerged(t *testing.T) {
 	abortErr := kvpb.NewTransactionAbortedError(kvpb.ABORT_REASON_ABORTED_RECORD_FOUND)
 	conditionFailedErr := &kvpb.ConditionFailedError{}
 	writeIntentErr := &kvpb.WriteIntentError{}
-	sendErr := sendError{}
+	sendErr := &sendError{}
 	ambiguousErr := &kvpb.AmbiguousResultError{}
 	randomErr := &kvpb.IntegerOverflowError{}
 
@@ -4437,7 +4437,7 @@ func TestEvictionTokenCoalesce(t *testing.T) {
 			// Return a sendError so DistSender retries the first range lookup in the
 			// user key-space for both batches.
 			if previouslyWaited := waitForInitialPuts(); !previouslyWaited {
-				return nil, newSendError("boom")
+				return nil, TestNewSendError("boom")
 			}
 			return br, nil
 		}
@@ -4742,7 +4742,7 @@ func TestRequestSubdivisionAfterDescriptorChangeWithUnavailableReplicasTerminate
 	transportFn := func(_ context.Context, ba *kvpb.BatchRequest) (*kvpb.BatchResponse, error) {
 		atomic.AddInt32(&numAttempts, 1)
 		require.Equal(t, 1, len(ba.Requests))
-		return nil, newSendError("boom")
+		return nil, TestNewSendError("boom")
 	}
 	rpcRetryOptions := &retry.Options{
 		MaxRetries: 5, // maxAttempts = 6
@@ -5167,7 +5167,7 @@ func TestSendToReplicasSkipsStaleReplicas(t *testing.T) {
 				get.Key = roachpb.Key("a")
 				ba.Add(get)
 				_, err = ds.sendToReplicas(ctx, ba, tok, false /* withCommit */)
-				require.IsType(t, sendError{}, err)
+				require.IsType(t, &sendError{}, err)
 				require.Regexp(t, "NotLeaseHolderError", err)
 				cached := rc.GetCached(ctx, tc.initialDesc.StartKey, false /* inverted */)
 				require.NotNil(t, cached)

pkg/kv/kvclient/kvcoord/replica_slice.go

@@ -12,14 +12,14 @@ package kvcoord
 
 import (
 	"context"
-	"fmt"
 	"sort"
 	"time"
 
 	"github.com/cockroachdb/cockroach/pkg/roachpb"
 	"github.com/cockroachdb/cockroach/pkg/settings/cluster"
 	"github.com/cockroachdb/cockroach/pkg/util/log"
 	"github.com/cockroachdb/cockroach/pkg/util/shuffle"
+	"github.com/cockroachdb/errors"
 )
 
 // ReplicaInfo extends the Replica structure with the associated node
@@ -135,7 +135,7 @@ func NewReplicaSlice(
 	}
 	if len(rs) == 0 {
 		return nil, newSendError(
-			fmt.Sprintf("no replica node information available via gossip for r%d", desc.RangeID))
+			errors.Errorf("no replica node information available via gossip for r%d", desc.RangeID))
 	}
 	return rs, nil
 }