kvpb: add a new protocol error ProxyFailedError

A ProxyFailedError can be returned in a BatchResponse if a proxy request
fails with a send error. The originator of the proxy request can use
this error to decide how to proceed. This error is necessary to prevent
cases where a network connection is broken mid-stream and the client may
need to perform different behavior based on whether the server received
the message.

Epic: none

Release note: None

docs/generated/metrics/metrics.html

@@ -894,6 +894,7 @@
 <tr><td>APPLICATION</td><td>distsender.rpc.err.notleaseholdererrtype</td><td>Number of NotLeaseHolderErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.oprequirestxnerrtype</td><td>Number of OpRequiresTxnErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.optimisticevalconflictserrtype</td><td>Number of OptimisticEvalConflictsErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
+<tr><td>APPLICATION</td><td>distsender.rpc.err.proxyfailederrtype</td><td>Number of ProxyFailedErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.raftgroupdeletederrtype</td><td>Number of RaftGroupDeletedErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.rangefeedretryerrtype</td><td>Number of RangeFeedRetryErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>
 <tr><td>APPLICATION</td><td>distsender.rpc.err.rangekeymismatcherrtype</td><td>Number of RangeKeyMismatchErrType errors received replica-bound RPCs<br/><br/>This counts how often error of the specified type was received back from replicas<br/>as part of executing possibly range-spanning requests. Failures to reach the target<br/>replica will be accounted for as &#39;roachpb.CommunicationErrType&#39; and unclassified<br/>errors as &#39;roachpb.InternalErrType&#39;.<br/></td><td>Errors</td><td>COUNTER</td><td>COUNT</td><td>AVG</td><td>NON_NEGATIVE_DERIVATIVE</td></tr>

pkg/kv/kvpb/errors.go

@@ -299,6 +299,7 @@ const (
 	MVCCHistoryMutationErrType              ErrorDetailType = 44
 	LockConflictErrType                     ErrorDetailType = 45
 	ReplicaUnavailableErrType               ErrorDetailType = 46
+	ProxyFailedErrType                      ErrorDetailType = 47
 	// When adding new error types, don't forget to update NumErrors below.
 
 	// CommunicationErrType indicates a gRPC error; this is not an ErrorDetail.
@@ -308,7 +309,7 @@ const (
 	// detail. The value 25 is chosen because it's reserved in the errors proto.
 	InternalErrType ErrorDetailType = 25
 
-	NumErrors int = 47
+	NumErrors int = 48
 )
 
 // Register the migration of all errors that used to be in the roachpb package
@@ -1711,6 +1712,56 @@ func (e *ReplicaUnavailableError) Type() ErrorDetailType {
 
 var _ ErrorDetailInterface = &ReplicaUnavailableError{}
 
+// Type is part of the ErrorDetailInterface.
+func (e *ProxyFailedError) Type() ErrorDetailType {
+	return ProxyFailedErrType
+}
+
+// Error is part of the builtin err interface
+func (e *ProxyFailedError) Error() string {
+	return redact.Sprint(e).StripMarkers()
+}
+
+// Format implements fmt.Formatter.
+func (e *ProxyFailedError) Format(s fmt.State, verb rune) { errors.FormatError(e, s, verb) }
+
+// SafeFormatError is part of the SafeFormatter
+func (e *ProxyFailedError) SafeFormatError(p errors.Printer) (next error) {
+	p.Printf("proxy failed with send error")
+	return nil
+}
+
+// Unwrap implements errors.Wrapper.
+func (e *ProxyFailedError) Unwrap() error {
+	return errors.DecodeError(context.Background(), e.Cause)
+}
+
+// NewProxyFailedError returns an ProxyFailedError wrapping (via
+// errors.Wrapper) the supplied error.
+func NewProxyFailedError(err error) *ProxyFailedError {
+	return &ProxyFailedError{
+		Cause: errors.EncodeError(context.Background(), err),
+	}
+}
+
+var _ ErrorDetailInterface = &ProxyFailedError{}
+var _ errors.SafeFormatter = (*ProxyFailedError)(nil)
+var _ fmt.Formatter = (*ProxyFailedError)(nil)
+var _ errors.Wrapper = (*ProxyFailedError)(nil)
+
+func init() {
+	encode := func(ctx context.Context, err error) (msgPrefix string, safeDetails []string, payload proto.Message) {
+		errors.As(err, &payload) // payload = err.(proto.Message)
+		return "", nil, payload
+	}
+	decode := func(ctx context.Context, cause error, msgPrefix string, safeDetails []string, payload proto.Message) error {
+		return payload.(*ProxyFailedError)
+	}
+	typeName := errors.GetTypeKey((*ProxyFailedError)(nil))
+	errors.RegisterWrapperEncoder(typeName, encode)
+	errors.RegisterWrapperDecoder(typeName, decode)
+}
+
 func init() {
 	errors.RegisterLeafDecoder(errors.GetTypeKey((*MissingRecordError)(nil)), func(_ context.Context, _ string, _ []string, _ proto.Message) error {
 		return &MissingRecordError{}
@@ -1755,3 +1806,4 @@ var _ errors.SafeFormatter = &RefreshFailedError{}
 var _ errors.SafeFormatter = &MVCCHistoryMutationError{}
 var _ errors.SafeFormatter = &UnhandledRetryableError{}
 var _ errors.SafeFormatter = &ReplicaUnavailableError{}
+var _ errors.SafeFormatter = &ProxyFailedError{}

pkg/kv/kvpb/errors.proto

@@ -418,6 +418,17 @@ message AmbiguousResultError {
   reserved 2;
 }
 
+// A ProxyFailedError is used to transmit a send error over the wire between a
+// proxy node and its final destination. The originator needs to handle
+// different types of SendErrors differently depending on both the type of the
+// error (specifically whether the error happened after the request started but
+// before we received a response) and the type of request (specifically whether
+// the request included a commit).
+message ProxyFailedError {
+  // The error that caused the proxy failure.
+  optional errorspb.EncodedError cause = 1 [(gogoproto.nullable) = false];
+}
+
 message ReplicaUnavailableError {
 
   optional roachpb.RangeDescriptor desc = 2 [(gogoproto.nullable) = false];

pkg/kv/kvpb/errors_test.go

@@ -15,6 +15,7 @@ import (
 	"context"
 	"fmt"
 	"io"
+	"reflect"
 	"strings"
 	"testing"
 
@@ -421,3 +422,28 @@ func TestDescNotFoundError(t *testing.T) {
 		require.True(t, errors.HasType(err, &DescNotFoundError{}))
 	})
 }
+
+// TestProxyFailedError validates that ProxyFailedErrors can be cleanly encoded
+// and decoded with an internal error.
+func TestProxyFailedError(t *testing.T) {
+	ctx := context.Background()
+	fooErr := errors.New("foo")
+	err := NewProxyFailedError(fooErr)
+	require.Equal(t, `proxy failed with send error`, err.Error())
+	require.True(t, errors.HasType(err, &ProxyFailedError{}))
+	decodedErr := errors.DecodeError(ctx, errors.EncodeError(ctx, err))
+
+	require.Truef(t, errors.HasType(decodedErr, &ProxyFailedError{}), "wrong error %v %v", decodedErr, reflect.TypeOf(decodedErr))
+	require.True(t, errors.Is(decodedErr, fooErr))
+	require.Equal(t, `proxy failed with send error`, decodedErr.Error())
+
+	var rue *ProxyFailedError
+	require.True(t, errors.As(decodedErr, &rue))
+
+	internalErr := errors.DecodeError(context.Background(), rue.Cause)
+	require.True(t, rue.Cause.IsSet())
+	require.ErrorContains(t, internalErr, "foo")
+	require.True(t, errors.Is(internalErr, fooErr))
+
+	require.Equal(t, `foo`, string(redact.Sprint(internalErr).Redact()))
+}