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 !isTransientConnectError(err) || ctx.Err() != nil {
106 return nil, err
107 }
108 continue
109 }
110 if isTransientHTTPStatus(resp.StatusCode) {
111 // Drain a bit of the body so we can include it in the
112 // final error if every retry exhausts. We cap the read
113 // at 4 KiB because edge proxies sometimes send pages.
114 body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
115 resp.Body.Close()
116 trimmed := strings.TrimSpace(string(body))
117 if resp.StatusCode == http.StatusTooManyRequests && isTerminalRateLimitBody(trimmed) {
118 // Subscription/quota/billing limits are not transient
119 // throttles; retrying only burns time. Surface the
120 // response immediately so the caller formats the
121 // normal "http 429: body" error.
122 return synthesizeResponse(resp.StatusCode, body), nil
123 }
124 lastErr = &transientHTTPError{Status: resp.StatusCode, Body: trimmed}
125 if attempt == streamRetryAttempts {
126 // Wrap as a real *http.Response shape the caller
127 // expects so it formats the error identically to a
128 // non-retried failure.
129 return synthesizeResponse(resp.StatusCode, body), nil
130 }
131 if d, ok := retryAfterDelay(resp.Header); ok {
132 if d > maxServerRetryDelay {
133 return nil, fmt.Errorf("server requested %s retry delay (max %s): http %d: %s",
134 d.Round(time.Second), maxServerRetryDelay, resp.StatusCode, trimmed)
135 }
136 serverDelay = d
137 }
138 continue
139 }
140 return resp, nil
141 }
142 if lastErr == nil {
143 lastErr = errors.New("retry loop exhausted")
144 }
145 return nil, lastErr
146}
147
148// transientHTTPError is the placeholder error returned while we're
149// still inside the retry loop. Callers never observe it directly;
150// the loop converts the final exhausted attempt into a synthesized
151// http.Response so the caller's existing "http NNN: body" formatting
152// keeps working unchanged.
153type transientHTTPError struct {
154 Status int
155 Body string
156}
157
158func (e *transientHTTPError) Error() string { return "transient http error" }
159
160// synthesizeResponse rebuilds a closable response wrapping the captured
161// body bytes so the caller's existing non-200 handling path keeps
162// working without needing to know retries happened.
163func synthesizeResponse(status int, body []byte) *http.Response {
164 return &http.Response{
165 StatusCode: status,
166 Body: io.NopCloser(bytes.NewReader(body)),
167 }
168}
169
170// isTransientConnectError reports whether err looks like a transient
171// network failure that's worth retrying. The classic "upstream connect
172// error or disconnect/reset before headers" from edge proxies (Envoy,
173// Cloudflare, GFE) shows up as several different concrete error types
174// across HTTP/1, HTTP/2, and TLS handshakes — easier to match by
175// substring on the rendered message than to enumerate every type.
176func isTransientConnectError(err error) bool {
177 if err == nil {
178 return false
179 }
180 if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
181 return false
182 }
183 msg := strings.ToLower(err.Error())
184 return strings.Contains(msg, "connection reset") ||
185 strings.Contains(msg, "connection refused") ||
186 strings.Contains(msg, "broken pipe") ||
187 strings.Contains(msg, "unexpected eof") ||
188 strings.Contains(msg, "eof") ||
189 strings.Contains(msg, "no such host") ||
190 strings.Contains(msg, "tls handshake") ||
191 strings.Contains(msg, "i/o timeout") ||
192 strings.Contains(msg, "transport failure") ||
193 strings.Contains(msg, "upstream connect error") ||
194 strings.Contains(msg, "disconnect/reset before headers")
195}
196
197// isTransientHTTPStatus reports whether a non-200 status code is
198// likely transient. 429 covers rate limits and capacity throttles
199// (terminal quota/billing 429s are filtered by isTerminalRateLimitBody
200// in the retry loop). 500 shows up for transient backend failures on
201// several providers, 524 is Cloudflare's origin timeout, and 529 is
202// the overloaded status used by Anthropic-style backends.
203func isTransientHTTPStatus(code int) bool {
204 switch code {
205 case http.StatusTooManyRequests,
206 http.StatusInternalServerError,
207 http.StatusBadGateway,
208 http.StatusServiceUnavailable,
209 http.StatusGatewayTimeout,
210 524, // Cloudflare: a timeout occurred
211 529: // overloaded
212 return true
213 }
214 return false
215}
216
217// isTerminalRateLimitBody reports whether a 429 body describes a
218// subscription, quota, or billing limit rather than a transient
219// throttle. Retrying those only delays the inevitable error, and on
220// some backends it can even extend the block.
221func isTerminalRateLimitBody(body string) bool {
222 msg := strings.ToLower(body)
223 for _, needle := range []string{
224 "gousagelimiterror", "freeusagelimiterror",
225 "usage limit", "monthly usage limit",
226 "available balance", "insufficient_quota",
227 "out of budget", "quota exceeded", "billing",
228 } {
229 if strings.Contains(msg, needle) {
230 return true
231 }
232 }
233 return false
234}
235
236// retryAfterDelay extracts a server-requested retry delay from
237// response headers. "retry-after-ms" (milliseconds) wins over the
238// standard "Retry-After", which may hold either delay-seconds or an
239// HTTP-date. Returns false when no parseable delay is present.
240func retryAfterDelay(h http.Header) (time.Duration, bool) {
241 if v := h.Get("retry-after-ms"); v != "" {
242 if ms, err := strconv.ParseFloat(strings.TrimSpace(v), 64); err == nil {
243 if ms < 0 {
244 ms = 0
245 }
246 return time.Duration(ms * float64(time.Millisecond)), true
247 }
248 }
249 v := h.Get("Retry-After")
250 if v == "" {
251 return 0, false
252 }
253 v = strings.TrimSpace(v)
254 if secs, err := strconv.ParseFloat(v, 64); err == nil {
255 if secs < 0 {
256 secs = 0
257 }
258 return time.Duration(secs * float64(time.Second)), true
259 }
260 if when, err := http.ParseTime(v); err == nil {
261 d := time.Until(when)
262 if d < 0 {
263 d = 0
264 }
265 return d, true
266 }
267 return 0, false
268}