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