1package provider
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "net/http"
10 "strconv"
11 "strings"
12 "time"
13
14 "github.com/google/uuid"
15)
16
17// requestMaxRetries is the maximum number of retries for transient connect or
18// HTTP failures before a streaming response opens. It matches Codex's default
19// request retry guard and remains separate from agent-level stream reconnects.
20const requestMaxRetries = 4
21
22// maxServerRetryDelay caps how long we are willing to honor a
23// server-requested Retry-After delay. When a provider asks for a
24// longer wait (e.g. a quota that resets in hours), the request fails
25// immediately with an informative error instead of silently blocking.
26const maxServerRetryDelay = 60 * time.Second
27
28// streamRetryBackoff returns the wait duration before retry attempt n
29// (1-based). Short, fixed backoff: 250ms, then 750ms for later retries.
30// Anything longer would feel like the agent froze; anything shorter starts
31// hammering the provider when it's actually struggling.
32func streamRetryBackoff(attempt int) time.Duration {
33 switch attempt {
34 case 1:
35 return 250 * time.Millisecond
36 default:
37 return 750 * time.Millisecond
38 }
39}
40
41// doStreamWithRetry performs an HTTP request that begins a streaming
42// response, with up to requestMaxRetries silent retries for transient
43// connect failures. Successful responses (status 200) and non-transient
44// failures (4xx, malformed bodies, etc.) are returned immediately.
45//
46// Retries fire on:
47// - net.Conn dial errors (connection refused, no such host, EOF
48// before headers, "upstream connect error", "connection reset")
49// - HTTP 429/500/502/503/504/524/529, except 429 responses whose
50// body indicates a terminal quota/billing limit
51//
52// When a retryable response carries a Retry-After (or retry-after-ms)
53// header, the wait before the next attempt uses the server-requested
54// delay instead of the fixed backoff, capped by maxServerRetryDelay.
55//
56// The handler is responsible for re-reading the request body each
57// attempt; we re-create the request via newReq() because http.Request
58// bodies are single-use. Callers in this package always Marshal a JSON
59// body up-front, so newReq just rebuilds with bytes.NewReader on the
60// captured payload.
61//
62// If every attempt fails, the last error/response is returned so the
63// caller can format a normal error message. On context cancellation
64// the loop bails out immediately with ctx.Err().
65func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() (*http.Request, error), lifecycle RequestLifecycle) (*http.Response, error) {
66 var lastErr error
67 serverDelay := time.Duration(-1) // <0: use the default fixed backoff
68 maxAttempts := requestMaxRetries + 1
69 for attempt := 0; attempt <= requestMaxRetries; attempt++ {
70 if attempt > 0 {
71 delay := streamRetryBackoff(attempt)
72 if serverDelay >= 0 {
73 delay = serverDelay
74 }
75 serverDelay = -1
76 if err := ctx.Err(); err != nil {
77 return nil, err
78 }
79 if lifecycle != nil {
80 lifecycle.RetryScheduled(attempt+1, maxAttempts, delay)
81 }
82 select {
83 case <-ctx.Done():
84 return nil, ctx.Err()
85 case <-time.After(delay):
86 }
87 }
88 if err := ctx.Err(); err != nil {
89 return nil, err
90 }
91 req, err := newReq()
92 if err != nil {
93 return nil, err
94 }
95 if err := ctx.Err(); err != nil {
96 return nil, err
97 }
98 reportRequestAttempt(lifecycle, attempt+1, maxAttempts)
99 resp, err := client.Do(req)
100 if err != nil {
101 lastErr = err
102 if ctx.Err() != nil {
103 return nil, err
104 }
105 transient := isTransientConnectError(err)
106 reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForError(err), !transient || attempt == requestMaxRetries)
107 if !transient {
108 return nil, err
109 }
110 continue
111 }
112 if isTransientHTTPStatus(resp.StatusCode) {
113 // Drain a bit of the body so we can include it in the
114 // final error if every retry exhausts. We cap the read
115 // at 4 KiB because edge proxies sometimes send pages.
116 body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
117 resp.Body.Close()
118 trimmed := strings.TrimSpace(string(body))
119 reason := requestFailureReasonForHTTP(resp.StatusCode, trimmed)
120 if resp.StatusCode == http.StatusTooManyRequests && isTerminalRateLimitBody(trimmed) {
121 // Subscription/quota/billing limits are not transient
122 // throttles; retrying only burns time. Surface the
123 // response immediately so the caller formats the
124 // normal "http 429: body" error.
125 reportRequestFailure(lifecycle, attempt+1, maxAttempts, RequestFailureQuota, true)
126 return synthesizeResponse(resp.StatusCode, body), nil
127 }
128 lastErr = &transientHTTPError{Status: resp.StatusCode, Body: trimmed}
129 if attempt == requestMaxRetries {
130 reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true)
131 // Wrap as a real *http.Response shape the caller
132 // expects so it formats the error identically to a
133 // non-retried failure.
134 return synthesizeResponse(resp.StatusCode, body), nil
135 }
136 if d, ok := retryAfterDelay(resp.Header); ok {
137 if d > maxServerRetryDelay {
138 reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, true)
139 return nil, fmt.Errorf("server requested %s retry delay (max %s): http %d: %s",
140 d.Round(time.Second), maxServerRetryDelay, resp.StatusCode, trimmed)
141 }
142 serverDelay = d
143 }
144 reportRequestFailure(lifecycle, attempt+1, maxAttempts, reason, false)
145 continue
146 }
147 if resp.StatusCode >= http.StatusBadRequest {
148 reportRequestFailure(lifecycle, attempt+1, maxAttempts, requestFailureReasonForHTTP(resp.StatusCode, ""), true)
149 }
150 return resp, nil
151 }
152 if lastErr == nil {
153 lastErr = errors.New("retry loop exhausted")
154 }
155 return nil, lastErr
156}
157
158func reportRequestAttempt(lifecycle RequestLifecycle, attempt, maxAttempts int) {
159 if lifecycle == nil {
160 return
161 }
162 lifecycle.RequestAttempt(attempt, maxAttempts)
163 if withID, ok := lifecycle.(RequestAttemptIDLifecycle); ok {
164 withID.RequestAttemptID(uuid.NewString(), attempt, maxAttempts)
165 }
166}
167
168// transientHTTPError is the placeholder error returned while we're
169// still inside the retry loop. Callers never observe it directly;
170// the loop converts the final exhausted attempt into a synthesized
171// http.Response so the caller's existing "http NNN: body" formatting
172// keeps working unchanged.
173type transientHTTPError struct {
174 Status int
175 Body string
176}
177
178func (e *transientHTTPError) Error() string { return "transient http error" }
179
180func reportRequestFailure(lifecycle RequestLifecycle, attempt, maxAttempts int, reason RequestFailureReason, terminal bool) {
181 if observer, ok := lifecycle.(RequestFailureLifecycle); ok {
182 observer.RequestFailed(attempt, maxAttempts, reason, terminal)
183 }
184}
185
186func requestFailureReasonForHTTP(status int, body string) RequestFailureReason {
187 if status >= http.StatusBadRequest && status < http.StatusInternalServerError {
188 if status == http.StatusTooManyRequests {
189 return RequestFailureRateLimit
190 }
191 return RequestFailureClient
192 }
193 msg := strings.ToLower(body)
194 if status == 529 || strings.Contains(msg, "overload") {
195 return RequestFailureOverload
196 }
197 return RequestFailureServer
198}
199
200func requestFailureReasonForError(err error) RequestFailureReason {
201 if err == nil {
202 return RequestFailureUnknown
203 }
204 if errors.Is(err, context.DeadlineExceeded) || strings.Contains(strings.ToLower(err.Error()), "timeout") {
205 return RequestFailureTimeout
206 }
207 if isTransientConnectError(err) {
208 return RequestFailureNetwork
209 }
210 return RequestFailureUnknown
211}
212
213// synthesizeResponse rebuilds a closable response wrapping the captured
214// body bytes so the caller's existing non-200 handling path keeps
215// working without needing to know retries happened.
216func synthesizeResponse(status int, body []byte) *http.Response {
217 return &http.Response{
218 StatusCode: status,
219 Body: io.NopCloser(bytes.NewReader(body)),
220 }
221}
222
223// isTransientConnectError reports whether err looks like a transient
224// network failure that's worth retrying. The classic "upstream connect
225// error or disconnect/reset before headers" from edge proxies (Envoy,
226// Cloudflare, GFE) shows up as several different concrete error types
227// across HTTP/1, HTTP/2, and TLS handshakes — easier to match by
228// substring on the rendered message than to enumerate every type.
229func isTransientConnectError(err error) bool {
230 if err == nil {
231 return false
232 }
233 if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
234 return false
235 }
236 msg := strings.ToLower(err.Error())
237 return strings.Contains(msg, "connection reset") ||
238 strings.Contains(msg, "connection refused") ||
239 strings.Contains(msg, "broken pipe") ||
240 strings.Contains(msg, "unexpected eof") ||
241 strings.Contains(msg, "eof") ||
242 strings.Contains(msg, "no such host") ||
243 strings.Contains(msg, "tls handshake") ||
244 strings.Contains(msg, "i/o timeout") ||
245 strings.Contains(msg, "transport failure") ||
246 strings.Contains(msg, "upstream connect error") ||
247 strings.Contains(msg, "disconnect/reset before headers")
248}
249
250// isTransientHTTPStatus reports whether a non-200 status code is
251// likely transient. 429 covers rate limits and capacity throttles
252// (terminal quota/billing 429s are filtered by isTerminalRateLimitBody
253// in the retry loop). 500 shows up for transient backend failures on
254// several providers, 524 is Cloudflare's origin timeout, and 529 is
255// the overloaded status used by Anthropic-style backends.
256func isTransientHTTPStatus(code int) bool {
257 switch code {
258 case http.StatusTooManyRequests,
259 http.StatusInternalServerError,
260 http.StatusBadGateway,
261 http.StatusServiceUnavailable,
262 http.StatusGatewayTimeout,
263 524, // Cloudflare: a timeout occurred
264 529: // overloaded
265 return true
266 }
267 return false
268}
269
270// isTerminalRateLimitBody reports whether a 429 body describes a
271// subscription, quota, or billing limit rather than a transient
272// throttle. Retrying those only delays the inevitable error, and on
273// some backends it can even extend the block.
274func isTerminalRateLimitBody(body string) bool {
275 msg := strings.ToLower(body)
276 for _, needle := range []string{
277 "gousagelimiterror", "freeusagelimiterror",
278 "usage limit", "monthly usage limit",
279 "available balance", "insufficient_quota",
280 "out of budget", "quota exceeded", "billing",
281 } {
282 if strings.Contains(msg, needle) {
283 return true
284 }
285 }
286 return false
287}
288
289// retryAfterDelay extracts a server-requested retry delay from
290// response headers. "retry-after-ms" (milliseconds) wins over the
291// standard "Retry-After", which may hold either delay-seconds or an
292// HTTP-date. Returns false when no parseable delay is present.
293func retryAfterDelay(h http.Header) (time.Duration, bool) {
294 if v := h.Get("retry-after-ms"); v != "" {
295 if ms, err := strconv.ParseFloat(strings.TrimSpace(v), 64); err == nil {
296 if ms < 0 {
297 ms = 0
298 }
299 return time.Duration(ms * float64(time.Millisecond)), true
300 }
301 }
302 v := h.Get("Retry-After")
303 if v == "" {
304 return 0, false
305 }
306 v = strings.TrimSpace(v)
307 if secs, err := strconv.ParseFloat(v, 64); err == nil {
308 if secs < 0 {
309 secs = 0
310 }
311 return time.Duration(secs * float64(time.Second)), true
312 }
313 if when, err := http.ParseTime(v); err == nil {
314 d := time.Until(when)
315 if d < 0 {
316 d = 0
317 }
318 return d, true
319 }
320 return 0, false
321}