diff --git a/pkg/connector/client/helpers.go b/pkg/connector/client/helpers.go index 1dc28a0d..3d4b2da2 100644 --- a/pkg/connector/client/helpers.go +++ b/pkg/connector/client/helpers.go @@ -57,10 +57,11 @@ func WrapError(err error, contextMsg string, annos *annotations.Annotations) err // for rate limit errors var slackLibrateLimitErr *slack.RateLimitedError if errors.As(err, &slackLibrateLimitErr) { + rlDesc := rateLimitDescription(slackLibrateLimitErr.RetryAfter) if annos != nil { - annos.WithRateLimiting(rateLimitDescription(slackLibrateLimitErr.RetryAfter)) + annos.WithRateLimiting(rlDesc) } - return uhttp.WrapErrors(codes.Unavailable, contextMsg, err) + return wrapErrorWithRateLimitDetails(codes.Unavailable, contextMsg, rlDesc, err) } // for 5xx status codes @@ -83,8 +84,12 @@ func WrapError(err error, contextMsg string, annos *annotations.Annotations) err } // Slack can return ok:false with "ratelimited" on HTTP 200. There's no // Retry-After header in this case, so use a default backoff. - if grpcCode == codes.Unavailable && annos != nil { - annos.WithRateLimiting(rateLimitDescription(defaultRateLimitRetryAfter)) + if grpcCode == codes.Unavailable { + rlDesc := rateLimitDescription(defaultRateLimitRetryAfter) + if annos != nil { + annos.WithRateLimiting(rlDesc) + } + return wrapErrorWithRateLimitDetails(grpcCode, contextMsg, rlDesc, err) } return uhttp.WrapErrors(grpcCode, contextMsg, err) } @@ -207,6 +212,26 @@ func RateLimitOverride() *v2.RateLimitDescription { return rateLimitDescription(60 * time.Second) } +// WrapErrorWithRateLimitOverride re-wraps an error with a 60s rate limit on +// both the gRPC status details (for SDK retry) and the annotations (for SyncOpResults). +func WrapErrorWithRateLimitOverride(err error, annos *annotations.Annotations) error { + rlDesc := RateLimitOverride() + if annos != nil { + annos.WithRateLimiting(rlDesc) + } + return wrapErrorWithRateLimitDetails(codes.Unavailable, "rate limited", rlDesc, err) +} + +// wrapErrorWithRateLimitDetails creates a gRPC error with RateLimitDescription +// attached as a status detail, so the SDK's retry logic knows how long to wait. +func wrapErrorWithRateLimitDetails(code codes.Code, msg string, rlDesc *v2.RateLimitDescription, err error) error { + st := status.New(code, msg) + if rlDesc != nil { + st, _ = st.WithDetails(rlDesc) + } + return errors.Join(st.Err(), err) +} + func rateLimitDescription(retryAfter time.Duration) *v2.RateLimitDescription { return &v2.RateLimitDescription{ Status: v2.RateLimitDescription_STATUS_OVERLIMIT, diff --git a/pkg/connector/client/helpers_test.go b/pkg/connector/client/helpers_test.go index c71924b6..779357de 100644 --- a/pkg/connector/client/helpers_test.go +++ b/pkg/connector/client/helpers_test.go @@ -7,6 +7,7 @@ import ( v2 "github.com/conductorone/baton-sdk/pb/c1/connector/v2" "github.com/conductorone/baton-sdk/pkg/annotations" "github.com/slack-go/slack" + "google.golang.org/grpc/status" ) func TestIsRateLimited(t *testing.T) { @@ -56,6 +57,31 @@ func TestWrapErrorSetsRateLimitAnnotation(t *testing.T) { } }) + t.Run("rate limit error attaches details to gRPC status", func(t *testing.T) { + err := &slack.RateLimitedError{RetryAfter: 4 * time.Second} + wrappedErr := WrapError(err, "test", nil) + st, ok := status.FromError(wrappedErr) + if !ok { + t.Fatal("expected gRPC status error") + } + found := false + for _, detail := range st.Details() { + if rl, ok := detail.(*v2.RateLimitDescription); ok { + found = true + if rl.Status != v2.RateLimitDescription_STATUS_OVERLIMIT { + t.Errorf("expected STATUS_OVERLIMIT, got %v", rl.Status) + } + resetIn := time.Until(rl.ResetAt.AsTime()) + if resetIn < 1*time.Second || resetIn > 5*time.Second { + t.Errorf("expected ResetAt ~4s from now, got %v", resetIn) + } + } + } + if !found { + t.Error("expected RateLimitDescription in gRPC status details") + } + }) + t.Run("non-rate-limit error does not populate rate limit annotation", func(t *testing.T) { var annos annotations.Annotations err := slack.SlackErrorResponse{Err: "user_not_found"} diff --git a/pkg/connector/user.go b/pkg/connector/user.go index 95861269..ba7ade88 100644 --- a/pkg/connector/user.go +++ b/pkg/connector/user.go @@ -246,7 +246,7 @@ func (o *userResourceType) listScimAPI(ctx context.Context, parentResourceID *v2 if err != nil { wrappedErr := client.WrapError(err, fmt.Sprintf("fetching user info for SCIM user %s", user.ID), &annos) if client.IsRateLimited(&annos) { - annos.WithRateLimiting(client.RateLimitOverride()) + wrappedErr = client.WrapErrorWithRateLimitOverride(wrappedErr, &annos) } return nil, &resource.SyncOpResults{Annotations: annos}, wrappedErr } diff --git a/pkg/connector/user_group.go b/pkg/connector/user_group.go index eb1b8630..9b66f1de 100644 --- a/pkg/connector/user_group.go +++ b/pkg/connector/user_group.go @@ -168,7 +168,7 @@ func (o *userGroupResourceType) Grants( if err != nil { wrappedErr := client.WrapError(err, fmt.Sprintf("fetching user info for member %s", member), &outputAnnotations) if client.IsRateLimited(&outputAnnotations) { - outputAnnotations.WithRateLimiting(client.RateLimitOverride()) + wrappedErr = client.WrapErrorWithRateLimitOverride(wrappedErr, &outputAnnotations) } return nil, &resource.SyncOpResults{Annotations: outputAnnotations}, wrappedErr }