grpc: introduce ErrRetriesExhausted to wrap retry failures (#8894)
Fixes: https://github.com/grpc/grpc-go/issues/7023
This change wraps RPC errors when the maximum retry limit is reached.
RELEASE NOTES:
* grpc: Introduce `ErrRetriesExhausted` to wrap RPC failures after
maximum retries are reached.
diff --git a/status/status_ext_test.go b/status/status_ext_test.go
index 5ff7693..2598ca6 100644
--- a/status/status_ext_test.go
+++ b/status/status_ext_test.go
@@ -21,6 +21,7 @@
import (
"context"
"errors"
+ "fmt"
"reflect"
"strings"
"testing"
@@ -259,3 +260,58 @@
}
}
}
+
+func (s) TestFromError_Wrapped(t *testing.T) {
+ base := status.New(codes.Canceled, "inner canceled")
+ sWithDetails, err := base.WithDetails(&testpb.Empty{})
+ if err != nil {
+ t.Fatalf("WithDetails failed: %v", err)
+ }
+ innerErr := sWithDetails.Err()
+ mustStatus := func(message string) *status.Status {
+ st, err := status.New(codes.Canceled, message).WithDetails(&testpb.Empty{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ return st
+ }
+
+ testCases := []struct {
+ name string
+ err error
+ wantStatus *status.Status
+ }{
+ {
+ name: "direct_error",
+ err: innerErr,
+ wantStatus: mustStatus("inner canceled"),
+ },
+ {
+ name: "wrapped_error",
+ err: fmt.Errorf("wrapped: %w", innerErr),
+ wantStatus: mustStatus("wrapped: rpc error: code = Canceled desc = inner canceled"),
+ },
+ {
+ name: "double_wrapped_error",
+ err: fmt.Errorf("outer: %w", fmt.Errorf("inner: %w", innerErr)),
+ wantStatus: mustStatus("outer: inner: rpc error: code = Canceled desc = inner canceled"),
+ },
+ {
+ name: "double_wrapped_single_errorf",
+ err: fmt.Errorf("error: %w: %w", errors.New("test error"), innerErr),
+ wantStatus: mustStatus("error: test error: rpc error: code = Canceled desc = inner canceled"),
+ },
+ }
+
+ for _, tc := range testCases {
+ t.Run(tc.name, func(t *testing.T) {
+ got, ok := status.FromError(tc.err)
+ if !ok {
+ t.Fatalf("status.FromError(%v) returned false; want true", tc.err)
+ }
+ if diff := cmp.Diff(tc.wantStatus, got, protocmp.Transform(), cmp.AllowUnexported(status.Status{})); diff != "" {
+ t.Fatalf("status.FromError(%v) got unexpected output, diff (-want +got):\n%s", tc.err, diff)
+ }
+ })
+ }
+}
diff --git a/stream.go b/stream.go
index eedb5f9..046f549 100644
--- a/stream.go
+++ b/stream.go
@@ -21,6 +21,7 @@
import (
"context"
"errors"
+ "fmt"
"io"
"math"
rand "math/rand/v2"
@@ -147,6 +148,15 @@
RecvMsg(m any) error
}
+// ErrRetriesExhausted is returned when an RPC exceeds its configured maximum
+// number of retry attempts.
+//
+// # Experimental
+//
+// Notice: This type is EXPERIMENTAL and may be changed or removed in a
+// later release.
+var ErrRetriesExhausted = errors.New("max retry attempts exhausted")
+
// NewStream creates a new Stream for the client side. This is typically
// called by generated code. ctx is used for the lifetime of the stream.
//
@@ -749,7 +759,7 @@
return false, err
}
if cs.numRetries+1 >= rp.MaxAttempts {
- return false, err
+ return false, fmt.Errorf("stopped after %d attempts: %w: %w", cs.numRetries+1, ErrRetriesExhausted, err)
}
var dur time.Duration
diff --git a/test/retry_test.go b/test/retry_test.go
index 72d79a5..af97be2 100644
--- a/test/retry_test.go
+++ b/test/retry_test.go
@@ -20,6 +20,7 @@
import (
"context"
+ "errors"
"fmt"
"io"
"net"
@@ -275,15 +276,15 @@
}
cErr := func(c codes.Code) clientOp {
return func(stream testgrpc.TestService_FullDuplexCallClient) error {
- want := status.New(c, "this is a test error").Err()
- if c == codes.OK {
- want = io.EOF
- }
res, err := stream.Recv()
- if res != nil ||
- ((err == nil) != (want == nil)) ||
- (want != nil && err.Error() != want.Error()) {
- return fmt.Errorf("client: Recv() = %v, %v; want <nil>, %v", res, err, want)
+ var gotCode codes.Code
+ if err == io.EOF {
+ gotCode = codes.OK
+ } else {
+ gotCode = status.Code(err)
+ }
+ if res != nil || gotCode != c {
+ return fmt.Errorf("client: Recv() = %v, %v; want <nil>, %v", res, err, c)
}
return nil
}
@@ -550,7 +551,10 @@
t.Fatalf("client: Recv() = %s, %v; want <nil>, error", got, err)
} else if status.Code(err) != codes.Unavailable {
t.Fatalf("client: Recv() = _, %v; want _, Unavailable", err)
+ } else if !errors.Is(err, grpc.ErrRetriesExhausted) {
+ t.Fatalf("want: ErrRetriesExhausted, got: %v", err)
}
+
if streamCallCount != tc.expectedAttempts {
t.Fatalf("stream expectedAttempts = %v; want %v", streamCallCount, tc.expectedAttempts)
}
@@ -816,3 +820,81 @@
stream1.Recv()
stream1.Send(&testpb.StreamingOutputCallRequest{})
}
+
+func (s) TestNoRetry(t *testing.T) {
+ scJSON := `{
+ "methodConfig": [{
+ "name": [{"service": "grpc.testing.TestService"}],
+ "retryPolicy": {
+ "MaxAttempts": 4,
+ "InitialBackoff": ".01s",
+ "MaxBackoff": ".01s",
+ "BackoffMultiplier": 1.0,
+ "RetryableStatusCodes": [ "UNAVAILABLE" ]
+ }
+ }]}`
+ tests := []struct {
+ name string
+ dialOpts []grpc.DialOption
+ }{
+ {
+ name: "disabled",
+ dialOpts: []grpc.DialOption{
+ grpc.WithDefaultServiceConfig(scJSON),
+ grpc.WithDisableRetry(),
+ },
+ },
+ {
+ name: "not_configured",
+ dialOpts: []grpc.DialOption{},
+ },
+ }
+
+ for _, tc := range tests {
+ t.Run(tc.name, func(t *testing.T) {
+ ss := &stubserver.StubServer{
+ FullDuplexCallF: func(testgrpc.TestService_FullDuplexCallServer) error {
+ return status.New(codes.Unavailable, "retryable error").Err()
+ },
+ EmptyCallF: func(context.Context, *testpb.Empty) (r *testpb.Empty, err error) {
+ return nil, status.New(codes.Unavailable, "retryable error").Err()
+ },
+ }
+ if err := ss.Start([]grpc.ServerOption{}, tc.dialOpts...); err != nil {
+ t.Fatalf("Error starting endpoint server: %v", err)
+ }
+ defer ss.Stop()
+
+ ctx, cancel := context.WithTimeout(context.Background(), defaultTestTimeout)
+ defer cancel()
+
+ // Test streaming RPC
+ stream, err := ss.Client.FullDuplexCall(ctx)
+ if err != nil {
+ t.Fatalf("Error while creating stream: %v", err)
+ }
+ _, err = stream.Recv()
+ if err == nil {
+ t.Fatal("stream.Recv() succeeded when expected to fail")
+ }
+ if status.Code(err) != codes.Unavailable {
+ t.Fatalf("client: Recv() = _, %v; want _, Unavailable", err)
+ }
+ if errors.Is(err, grpc.ErrRetriesExhausted) {
+ t.Fatalf("client: Recv() error matches ErrRetriesExhausted, want not match")
+ }
+
+ // Test unary RPC
+ _, err = ss.Client.EmptyCall(ctx, &testpb.Empty{})
+ if err == nil {
+ t.Fatal("EmptyCall() succeeded when expected to fail")
+ }
+ if status.Code(err) != codes.Unavailable {
+ t.Fatalf("client: EmptyCall() = _, %v; want _, Unavailable", err)
+ }
+ if errors.Is(err, grpc.ErrRetriesExhausted) {
+ t.Fatalf("client: EmptyCall() error matches ErrRetriesExhausted, want not match")
+ }
+ })
+ }
+}
