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