Stop the retry backoff from outliving its deadline
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>
This commit is contained in:
@@ -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
|
||||
}
|
||||
|
||||
@@ -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<<attempt)) * time.Millisecond
|
||||
select {
|
||||
case <-req.Context().Done():
|
||||
return nil, req.Context().Err()
|
||||
case <-time.After(backoff):
|
||||
}
|
||||
}
|
||||
|
||||
return lastResp, nil
|
||||
}
|
||||
|
||||
func (d *GoogleWorkspaceDriver) ListAccounts(ctx context.Context) ([]AccountRecord, error) {
|
||||
adminService, err := admin.NewService(ctx, option.WithHTTPClient(d.httpClient))
|
||||
if err != nil {
|
||||
|
||||
288
pkg/accessreview/drivers/retry_test.go
Normal file
288
pkg/accessreview/drivers/retry_test.go
Normal file
@@ -0,0 +1,288 @@
|
||||
// Copyright (c) 2026 Probo Inc <hello@probo.com>.
|
||||
//
|
||||
// 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)
|
||||
}
|
||||
})
|
||||
}
|
||||
Reference in New Issue
Block a user