1package provider
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "io"
8 "net/http"
9 "strings"
10 "time"
11)
12
13// streamRetryAttempts is the maximum number of silent retries for
14// transient connect failures before the streaming request is opened.
15// 2 retries (so up to 3 total attempts) covers the common case where
16// the upstream provider's edge proxy briefly returns 502/503/504 or
17// resets the connection before headers, without making the user wait
18// noticeably long if the outage is real.
19const streamRetryAttempts = 2
20
21// streamRetryBackoff returns the wait duration before retry attempt n
22// (1-based). Short, fixed backoff: 250ms, then 750ms. Anything longer
23// would feel like the agent froze; anything shorter starts hammering
24// the provider when it's actually struggling.
25func streamRetryBackoff(attempt int) time.Duration {
26 switch attempt {
27 case 1:
28 return 250 * time.Millisecond
29 default:
30 return 750 * time.Millisecond
31 }
32}
33
34// doStreamWithRetry performs an HTTP request that begins a streaming
35// response, with up to streamRetryAttempts silent retries for transient
36// connect failures. Successful responses (status 200) and non-transient
37// failures (4xx, malformed bodies, etc.) are returned immediately.
38//
39// Retries fire on:
40// - net.Conn dial errors (connection refused, no such host, EOF
41// before headers, "upstream connect error", "connection reset")
42// - HTTP 502/503/504
43//
44// The handler is responsible for re-reading the request body each
45// attempt; we re-create the request via newReq() because http.Request
46// bodies are single-use. Callers in this package always Marshal a JSON
47// body up-front, so newReq just rebuilds with bytes.NewReader on the
48// captured payload.
49//
50// If every attempt fails, the last error/response is returned so the
51// caller can format a normal error message. On context cancellation
52// the loop bails out immediately with ctx.Err().
53func doStreamWithRetry(ctx context.Context, client *http.Client, newReq func() (*http.Request, error)) (*http.Response, error) {
54 var lastErr error
55 for attempt := 0; attempt <= streamRetryAttempts; attempt++ {
56 if attempt > 0 {
57 select {
58 case <-ctx.Done():
59 return nil, ctx.Err()
60 case <-time.After(streamRetryBackoff(attempt)):
61 }
62 }
63 req, err := newReq()
64 if err != nil {
65 return nil, err
66 }
67 resp, err := client.Do(req)
68 if err != nil {
69 lastErr = err
70 if !isTransientConnectError(err) || ctx.Err() != nil {
71 return nil, err
72 }
73 continue
74 }
75 if isTransientHTTPStatus(resp.StatusCode) {
76 // Drain a bit of the body so we can include it in the
77 // final error if every retry exhausts. We cap the read
78 // at 4 KiB because edge proxies sometimes send pages.
79 body, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
80 resp.Body.Close()
81 lastErr = &transientHTTPError{Status: resp.StatusCode, Body: strings.TrimSpace(string(body))}
82 if attempt == streamRetryAttempts {
83 // Wrap as a real *http.Response shape the caller
84 // expects so it formats the error identically to a
85 // non-retried failure.
86 return synthesizeResponse(resp.StatusCode, body), nil
87 }
88 continue
89 }
90 return resp, nil
91 }
92 if lastErr == nil {
93 lastErr = errors.New("retry loop exhausted")
94 }
95 return nil, lastErr
96}
97
98// transientHTTPError is the placeholder error returned while we're
99// still inside the retry loop. Callers never observe it directly;
100// the loop converts the final exhausted attempt into a synthesized
101// http.Response so the caller's existing "http NNN: body" formatting
102// keeps working unchanged.
103type transientHTTPError struct {
104 Status int
105 Body string
106}
107
108func (e *transientHTTPError) Error() string { return "transient http error" }
109
110// synthesizeResponse rebuilds a closable response wrapping the captured
111// body bytes so the caller's existing non-200 handling path keeps
112// working without needing to know retries happened.
113func synthesizeResponse(status int, body []byte) *http.Response {
114 return &http.Response{
115 StatusCode: status,
116 Body: io.NopCloser(bytes.NewReader(body)),
117 }
118}
119
120// isTransientConnectError reports whether err looks like a transient
121// network failure that's worth retrying. The classic "upstream connect
122// error or disconnect/reset before headers" from edge proxies (Envoy,
123// Cloudflare, GFE) shows up as several different concrete error types
124// across HTTP/1, HTTP/2, and TLS handshakes — easier to match by
125// substring on the rendered message than to enumerate every type.
126func isTransientConnectError(err error) bool {
127 if err == nil {
128 return false
129 }
130 if errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
131 return false
132 }
133 msg := strings.ToLower(err.Error())
134 return strings.Contains(msg, "connection reset") ||
135 strings.Contains(msg, "connection refused") ||
136 strings.Contains(msg, "broken pipe") ||
137 strings.Contains(msg, "unexpected eof") ||
138 strings.Contains(msg, "eof") ||
139 strings.Contains(msg, "no such host") ||
140 strings.Contains(msg, "tls handshake") ||
141 strings.Contains(msg, "i/o timeout") ||
142 strings.Contains(msg, "transport failure") ||
143 strings.Contains(msg, "upstream connect error") ||
144 strings.Contains(msg, "disconnect/reset before headers")
145}
146
147// isTransientHTTPStatus reports whether a non-200 status code is
148// likely transient (502/503/504). 5xx outside this range is surfaced
149// immediately because it usually indicates a real backend bug.
150func isTransientHTTPStatus(code int) bool {
151 switch code {
152 case http.StatusBadGateway,
153 http.StatusServiceUnavailable,
154 http.StatusGatewayTimeout:
155 return true
156 }
157 return false
158}