From 92beb20dcc2d949a1f5408d502a222d5ce8b25ff Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Aur=C3=A9lien=20Sibiril?= <81782+aureliensibiril@users.noreply.github.com> Date: Sat, 25 Jul 2026 09:32:38 +0200 Subject: [PATCH] Stop the retry backoff from outliving its deadline MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A fetch runs under a 30-second per-source budget, but the retry transport slept without reference to it, so backoff could convert a reportable provider status into an opaque "context deadline exceeded". Three changes. The final attempt no longer sleeps: nothing follows it, so the wait only spent the caller's deadline to return a response already in hand — up to a second per failed request, across sixteen drivers. Retry-After is now honoured, in both the delta-seconds and HTTP-date forms; ignoring it meant retrying a 429 after 250ms and earning another 429, spending the whole retry budget in under a second. And a wait is skipped entirely when it exceeds the remaining deadline or a 5s cap, because a retry that lands after the deadline cannot succeed — the throttled response is surfaced instead so the caller reports what the provider actually said. The type moves from google_workspace.go to driver.go, which is where the other shared driver machinery lives; sixteen drivers construct it and none of them are Google Workspace. It had no tests, so it has them now. Signed-off-by: Aurélien Sibiril <81782+aureliensibiril@users.noreply.github.com> --- pkg/accessreview/drivers/driver.go | 111 +++++++ pkg/accessreview/drivers/google_workspace.go | 45 --- pkg/accessreview/drivers/retry_test.go | 288 +++++++++++++++++++ 3 files changed, 399 insertions(+), 45 deletions(-) create mode 100644 pkg/accessreview/drivers/retry_test.go diff --git a/pkg/accessreview/drivers/driver.go b/pkg/accessreview/drivers/driver.go index f04fc9d8d..df3a47fb5 100644 --- a/pkg/accessreview/drivers/driver.go +++ b/pkg/accessreview/drivers/driver.go @@ -21,8 +21,12 @@ package drivers import ( + "bytes" "context" "fmt" + "io" + "net/http" + "strconv" "strings" "time" @@ -137,3 +141,110 @@ func ownerMemberRoles(role string) []string { func isOwnerRole(role string) bool { return strings.EqualFold(strings.TrimSpace(role), "owner") } + +// retryRoundTripper retries 429 and 5xx responses with exponential backoff. +// +// The retry budget is bounded by what can still produce a useful answer: a +// fetch runs under a per-source deadline, so a sleep that outlives the +// deadline, or that follows the final attempt, only converts a reportable +// provider status into an opaque timeout. Every wait below is therefore +// guarded. +type retryRoundTripper struct { + next http.RoundTripper + maxRetries int +} + +const ( + // retryBaseBackoff is the first backoff step; it doubles per attempt. + retryBaseBackoff = 250 * time.Millisecond + // maxRetryWait caps a single wait. A provider asking for longer (via + // Retry-After) cannot be accommodated inside a per-source budget, so the + // throttled response is returned instead and the caller reports it. + maxRetryWait = 5 * time.Second +) + +func (rt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { + transport := rt.next + if transport == nil { + transport = http.DefaultTransport + } + + var lastResp *http.Response + + for attempt := range rt.maxRetries { + resp, err := transport.RoundTrip(req) + if err != nil { + return nil, err + } + + if resp.StatusCode != http.StatusTooManyRequests && resp.StatusCode < 500 { + return resp, nil + } + + // Buffer and re-attach the body so the caller can still read it + // if this turns out to be the final (retry-exhausted) response. + body, _ := io.ReadAll(resp.Body) + _ = resp.Body.Close() + resp.Body = io.NopCloser(bytes.NewReader(body)) + lastResp = resp + + // Nothing follows the last attempt, so waiting here would burn the + // caller's deadline to return the response it already has. + if attempt == rt.maxRetries-1 { + break + } + + wait := retryBaseBackoff << attempt + // Retry-After is authoritative on 429: retrying sooner just earns + // another 429 and spends an attempt doing it. + if after, ok := retryAfter(resp); ok { + wait = after + } + + if wait > maxRetryWait { + break + } + + // Sleeping past the deadline guarantees a context error that hides + // the provider's actual status from the caller. + if deadline, ok := req.Context().Deadline(); ok && time.Until(deadline) <= wait { + break + } + + timer := time.NewTimer(wait) + + select { + case <-req.Context().Done(): + timer.Stop() + + return nil, req.Context().Err() + case <-timer.C: + } + } + + return lastResp, nil +} + +// retryAfter reads a Retry-After header in either documented form — +// delta-seconds or an HTTP-date. The bool reports whether the header was +// present and parseable; a date already in the past yields a zero wait. +func retryAfter(resp *http.Response) (time.Duration, bool) { + value := strings.TrimSpace(resp.Header.Get("Retry-After")) + if value == "" { + return 0, false + } + + if seconds, err := strconv.Atoi(value); err == nil { + if seconds < 0 { + return 0, false + } + + return time.Duration(seconds) * time.Second, true + } + + if at, err := http.ParseTime(value); err == nil { + return max(time.Until(at), 0), true + } + + return 0, false +} diff --git a/pkg/accessreview/drivers/google_workspace.go b/pkg/accessreview/drivers/google_workspace.go index 1f685d278..79ecec7ff 100644 --- a/pkg/accessreview/drivers/google_workspace.go +++ b/pkg/accessreview/drivers/google_workspace.go @@ -21,10 +21,8 @@ package drivers import ( - "bytes" "context" "fmt" - "io" "net/http" "time" @@ -50,49 +48,6 @@ func NewGoogleWorkspaceDriver(httpClient *http.Client) *GoogleWorkspaceDriver { } } -// retryRoundTripper retries requests that receive 5xx or 429 responses -// with exponential backoff. -type retryRoundTripper struct { - next http.RoundTripper - maxRetries int -} - -func (rt *retryRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { - transport := rt.next - if transport == nil { - transport = http.DefaultTransport - } - - var lastResp *http.Response - - for attempt := range rt.maxRetries { - resp, err := transport.RoundTrip(req) - if err != nil { - return nil, err - } - - if resp.StatusCode != http.StatusTooManyRequests && resp.StatusCode < 500 { - return resp, nil - } - - // Buffer and re-attach the body so the caller can still read it - // if this turns out to be the final (retry-exhausted) response. - body, _ := io.ReadAll(resp.Body) - _ = resp.Body.Close() - resp.Body = io.NopCloser(bytes.NewReader(body)) - lastResp = resp - - backoff := time.Duration(250*(1<. +// +// Permission is hereby granted, free of charge, to any person obtaining a copy +// of this software and associated documentation files (the "Software"), to deal +// in the Software without restriction, including without limitation the rights +// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell +// copies of the Software, and to permit persons to whom the Software is +// furnished to do so, subject to the following conditions: +// +// The above copyright notice and this permission notice shall be included in +// all copies or substantial portions of the Software. +// +// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR +// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, +// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE +// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER +// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, +// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE +// SOFTWARE. + +package drivers + +import ( + "context" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// stubTransport replays a fixed sequence of responses and records how many +// requests it received. +type stubTransport struct { + responses []*http.Response + calls int +} + +func (s *stubTransport) RoundTrip(*http.Request) (*http.Response, error) { + resp := s.responses[min(s.calls, len(s.responses)-1)] + s.calls++ + + return resp, nil +} + +func response(status int, body string, header http.Header) *http.Response { + if header == nil { + header = http.Header{} + } + + return &http.Response{ + StatusCode: status, + Body: io.NopCloser(strings.NewReader(body)), + Header: header, + } +} + +func newRequest(t *testing.T, ctx context.Context) *http.Request { + t.Helper() + + req, err := http.NewRequestWithContext(ctx, http.MethodGet, "https://api.example.com/users", nil) + require.NoError(t, err) + + return req +} + +func TestRetryRoundTripper(t *testing.T) { + t.Parallel() + + t.Run("returns a success without retrying", func(t *testing.T) { + t.Parallel() + + stub := &stubTransport{responses: []*http.Response{response(http.StatusOK, `{"ok":true}`, nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, resp.StatusCode) + assert.Equal(t, 1, stub.calls) + }) + + t.Run("does not retry a client error", func(t *testing.T) { + t.Parallel() + + stub := &stubTransport{responses: []*http.Response{response(http.StatusForbidden, "denied", nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + require.NoError(t, err) + assert.Equal(t, http.StatusForbidden, resp.StatusCode) + assert.Equal(t, 1, stub.calls) + }) + + t.Run("retries a 5xx then succeeds", func(t *testing.T) { + t.Parallel() + + stub := &stubTransport{responses: []*http.Response{ + response(http.StatusBadGateway, "boom", nil), + response(http.StatusOK, `{"ok":true}`, nil), + }} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + require.NoError(t, err) + assert.Equal(t, http.StatusOK, resp.StatusCode) + assert.Equal(t, 2, stub.calls) + }) + + // The exhausted response must still be readable: callers decode the error + // body to report the provider's message. + t.Run("returns a readable body after exhausting retries", func(t *testing.T) { + t.Parallel() + + stub := &stubTransport{responses: []*http.Response{response(http.StatusInternalServerError, "upstream failed", nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 2} + + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + require.NoError(t, err) + assert.Equal(t, http.StatusInternalServerError, resp.StatusCode) + + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + assert.Equal(t, "upstream failed", string(body)) + }) + + // The final attempt has nothing after it, so the old code's trailing sleep + // only spent the caller's deadline to return a response it already had. + t.Run("does not sleep after the final attempt", func(t *testing.T) { + t.Parallel() + + stub := &stubTransport{responses: []*http.Response{response(http.StatusServiceUnavailable, "down", nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + start := time.Now() + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Equal(t, http.StatusServiceUnavailable, resp.StatusCode) + assert.Equal(t, 3, stub.calls) + // Two waits (250ms + 500ms), never a third after the last attempt. + assert.Less(t, elapsed, 900*time.Millisecond) + assert.GreaterOrEqual(t, elapsed, 750*time.Millisecond) + }) + + // A long Retry-After cannot be honoured inside a per-source budget, so the + // throttled response is surfaced immediately rather than slept through. + t.Run("returns immediately when Retry-After exceeds the cap", func(t *testing.T) { + t.Parallel() + + header := http.Header{} + header.Set("Retry-After", "60") + + stub := &stubTransport{responses: []*http.Response{response(http.StatusTooManyRequests, "slow down", header)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + start := time.Now() + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode) + assert.Equal(t, 1, stub.calls, "must not spend a second attempt it cannot wait for") + assert.Less(t, elapsed, 100*time.Millisecond) + }) + + // Retry-After shorter than the cap is honoured over the exponential step. + t.Run("honours a short Retry-After", func(t *testing.T) { + t.Parallel() + + header := http.Header{} + header.Set("Retry-After", "1") + + stub := &stubTransport{responses: []*http.Response{ + response(http.StatusTooManyRequests, "slow down", header), + response(http.StatusOK, `{"ok":true}`, nil), + }} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + start := time.Now() + resp, err := rt.RoundTrip(newRequest(t, t.Context())) + elapsed := time.Since(start) + + require.NoError(t, err) + assert.Equal(t, http.StatusOK, resp.StatusCode) + // 1s from Retry-After, not the 250ms exponential step. + assert.GreaterOrEqual(t, elapsed, 1*time.Second) + }) + + // Sleeping past the deadline turns a reportable 429 into an opaque + // "context deadline exceeded". + t.Run("returns the response rather than sleeping past the deadline", func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(t.Context(), 50*time.Millisecond) + defer cancel() + + stub := &stubTransport{responses: []*http.Response{response(http.StatusTooManyRequests, "slow down", nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + start := time.Now() + resp, err := rt.RoundTrip(newRequest(t, ctx)) + elapsed := time.Since(start) + + require.NoError(t, err) + require.NotNil(t, resp) + assert.Equal(t, http.StatusTooManyRequests, resp.StatusCode) + assert.Equal(t, 1, stub.calls) + assert.Less(t, elapsed, 50*time.Millisecond) + }) + + t.Run("aborts when the context is cancelled mid-backoff", func(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithCancel(t.Context()) + + stub := &stubTransport{responses: []*http.Response{response(http.StatusInternalServerError, "boom", nil)}} + rt := &retryRoundTripper{next: stub, maxRetries: 3} + + go func() { + time.Sleep(30 * time.Millisecond) + cancel() + }() + + _, err := rt.RoundTrip(newRequest(t, ctx)) + require.ErrorIs(t, err, context.Canceled) + }) +} + +func TestRetryAfter(t *testing.T) { + t.Parallel() + + t.Run("absent header", func(t *testing.T) { + t.Parallel() + + _, ok := retryAfter(response(http.StatusTooManyRequests, "", nil)) + assert.False(t, ok) + }) + + t.Run("delta seconds", func(t *testing.T) { + t.Parallel() + + header := http.Header{} + header.Set("Retry-After", "30") + + got, ok := retryAfter(response(http.StatusTooManyRequests, "", header)) + require.True(t, ok) + assert.Equal(t, 30*time.Second, got) + }) + + t.Run("http date in the future", func(t *testing.T) { + t.Parallel() + + header := http.Header{} + header.Set("Retry-After", time.Now().Add(2*time.Second).UTC().Format(http.TimeFormat)) + + got, ok := retryAfter(response(http.StatusTooManyRequests, "", header)) + require.True(t, ok) + assert.Positive(t, got) + assert.LessOrEqual(t, got, 2*time.Second) + }) + + t.Run("http date in the past yields no wait", func(t *testing.T) { + t.Parallel() + + header := http.Header{} + header.Set("Retry-After", time.Now().Add(-time.Hour).UTC().Format(http.TimeFormat)) + + got, ok := retryAfter(response(http.StatusTooManyRequests, "", header)) + require.True(t, ok) + assert.Zero(t, got) + }) + + t.Run("garbage and negative values are ignored", func(t *testing.T) { + t.Parallel() + + for _, value := range []string{"soon", "-5", " "} { + header := http.Header{} + header.Set("Retry-After", value) + + _, ok := retryAfter(response(http.StatusTooManyRequests, "", header)) + assert.False(t, ok, value) + } + }) +}