1package provider
2
3import (
4 "context"
5 "net/http"
6 "net/http/httptest"
7 "strings"
8 "sync/atomic"
9 "testing"
10 "time"
11)
12
13// TestOpenAIStreamRetriesOn503 verifies the OpenAI/Kimi stream client
14// silently retries when the upstream returns 503 a couple of times,
15// then succeeds. The user-visible error path should NOT fire because
16// a later attempt landed on a 200.
17func TestOpenAIStreamRetriesOn503(t *testing.T) {
18 var attempts atomic.Int32
19 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
20 n := attempts.Add(1)
21 if n < 3 {
22 w.WriteHeader(http.StatusServiceUnavailable)
23 _, _ = w.Write([]byte("upstream connect error"))
24 return
25 }
26 w.Header().Set("content-type", "text/event-stream")
27 w.WriteHeader(http.StatusOK)
28 fl, _ := w.(http.Flusher)
29 write := func(s string) {
30 _, _ = w.Write([]byte(s))
31 if fl != nil {
32 fl.Flush()
33 }
34 }
35 write("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n")
36 write("data: [DONE]\n\n")
37 }))
38 defer srv.Close()
39
40 c := NewOpenAI("x", srv.URL)
41 evs, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
42 if err != nil {
43 t.Fatalf("stream returned error: %v", err)
44 }
45 gotText := ""
46 for ev := range evs {
47 if td, ok := ev.(EventTextDelta); ok {
48 gotText += td.Delta
49 }
50 }
51 if gotText != "ok" {
52 t.Fatalf("text=%q", gotText)
53 }
54 if got := attempts.Load(); got != 3 {
55 t.Fatalf("attempts=%d want 3", got)
56 }
57}
58
59// TestOpenAIStreamSurfaces503AfterRetriesExhausted verifies that when
60// every retry attempt returns 503 the caller sees a normal "http 503"
61// error, not a retry-internal placeholder. The body bytes must be
62// preserved so the rescue picker / red banner can show them.
63func TestOpenAIStreamSurfaces503AfterRetriesExhausted(t *testing.T) {
64 var attempts atomic.Int32
65 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
66 attempts.Add(1)
67 w.WriteHeader(http.StatusServiceUnavailable)
68 _, _ = w.Write([]byte("upstream connect error"))
69 }))
70 defer srv.Close()
71
72 c := NewOpenAI("x", srv.URL)
73 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
74 if err == nil {
75 t.Fatalf("expected error after exhausted retries")
76 }
77 if !strings.Contains(err.Error(), "503") {
78 t.Fatalf("error %q should mention 503", err)
79 }
80 if got := attempts.Load(); got != streamRetryAttempts+1 {
81 t.Fatalf("attempts=%d want %d", got, streamRetryAttempts+1)
82 }
83}
84
85// TestOpenAIStreamDoesNotRetryOn400 ensures non-transient errors are
86// returned immediately without retrying. We don't want to amplify
87// load on a provider that's actively rejecting our request shape.
88func TestOpenAIStreamDoesNotRetryOn400(t *testing.T) {
89 var attempts atomic.Int32
90 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
91 attempts.Add(1)
92 w.WriteHeader(http.StatusBadRequest)
93 _, _ = w.Write([]byte(`{"error":{"message":"nope"}}`))
94 }))
95 defer srv.Close()
96
97 c := NewOpenAI("x", srv.URL)
98 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
99 if err == nil || !strings.Contains(err.Error(), "400") {
100 t.Fatalf("want 400 err got %v", err)
101 }
102 if got := attempts.Load(); got != 1 {
103 t.Fatalf("attempts=%d want 1", got)
104 }
105}
106
107func TestIsTransientHTTPStatus(t *testing.T) {
108 for _, code := range []int{429, 500, 502, 503, 504, 524, 529} {
109 if !isTransientHTTPStatus(code) {
110 t.Errorf("%d should be transient", code)
111 }
112 }
113 for _, code := range []int{200, 400, 401, 403, 404, 501} {
114 if isTransientHTTPStatus(code) {
115 t.Errorf("%d should NOT be transient", code)
116 }
117 }
118}
119
120// TestOpenAIStreamRetriesOn429WithRetryAfter verifies that a transient
121// 429 is retried and that a server-requested Retry-After delay is
122// honored instead of the fixed backoff.
123func TestOpenAIStreamRetriesOn429WithRetryAfter(t *testing.T) {
124 var attempts atomic.Int32
125 var firstRetryAt atomic.Int64
126 start := time.Now()
127 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
128 n := attempts.Add(1)
129 if n == 1 {
130 w.Header().Set("retry-after-ms", "50")
131 w.WriteHeader(http.StatusTooManyRequests)
132 _, _ = w.Write([]byte("Our servers are currently overloaded. Please try again later."))
133 return
134 }
135 firstRetryAt.Store(int64(time.Since(start)))
136 w.Header().Set("content-type", "text/event-stream")
137 w.WriteHeader(http.StatusOK)
138 fl, _ := w.(http.Flusher)
139 write := func(s string) {
140 _, _ = w.Write([]byte(s))
141 if fl != nil {
142 fl.Flush()
143 }
144 }
145 write("data: {\"choices\":[{\"index\":0,\"delta\":{\"content\":\"ok\"},\"finish_reason\":\"stop\"}]}\n\n")
146 write("data: [DONE]\n\n")
147 }))
148 defer srv.Close()
149
150 c := NewOpenAI("x", srv.URL)
151 evs, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
152 if err != nil {
153 t.Fatalf("stream returned error: %v", err)
154 }
155 for range evs {
156 }
157 if got := attempts.Load(); got != 2 {
158 t.Fatalf("attempts=%d want 2", got)
159 }
160 if got := time.Duration(firstRetryAt.Load()); got < 50*time.Millisecond {
161 t.Fatalf("retry fired after %v; want >= 50ms (server-requested delay)", got)
162 }
163}
164
165// TestOpenAIStreamDoesNotRetryTerminal429 ensures quota/billing 429s
166// are surfaced immediately without retrying: a usage-limit block will
167// not clear within the retry window, and hammering the endpoint can
168// extend the block on some backends.
169func TestOpenAIStreamDoesNotRetryTerminal429(t *testing.T) {
170 var attempts atomic.Int32
171 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
172 attempts.Add(1)
173 w.WriteHeader(http.StatusTooManyRequests)
174 _, _ = w.Write([]byte(`{"error":{"type":"insufficient_quota","message":"You exceeded your current quota"}}`))
175 }))
176 defer srv.Close()
177
178 c := NewOpenAI("x", srv.URL)
179 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
180 if err == nil || !strings.Contains(err.Error(), "429") {
181 t.Fatalf("want 429 err got %v", err)
182 }
183 if got := attempts.Load(); got != 1 {
184 t.Fatalf("attempts=%d want 1", got)
185 }
186}
187
188// TestOpenAIStreamRejectsExcessiveRetryAfter verifies that a
189// server-requested delay beyond maxServerRetryDelay fails immediately
190// with an informative error instead of blocking silently.
191func TestOpenAIStreamRejectsExcessiveRetryAfter(t *testing.T) {
192 var attempts atomic.Int32
193 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
194 attempts.Add(1)
195 w.Header().Set("Retry-After", "18000") // 5 hours
196 w.WriteHeader(http.StatusTooManyRequests)
197 _, _ = w.Write([]byte("rate limited"))
198 }))
199 defer srv.Close()
200
201 c := NewOpenAI("x", srv.URL)
202 _, err := c.Stream(context.Background(), Request{Model: "gpt-5"})
203 if err == nil || !strings.Contains(err.Error(), "retry delay") {
204 t.Fatalf("want retry-delay err got %v", err)
205 }
206 if got := attempts.Load(); got != 1 {
207 t.Fatalf("attempts=%d want 1", got)
208 }
209}
210
211func TestRetryAfterDelay(t *testing.T) {
212 mk := func(kv ...string) http.Header {
213 h := http.Header{}
214 for i := 0; i+1 < len(kv); i += 2 {
215 h.Set(kv[i], kv[i+1])
216 }
217 return h
218 }
219 if d, ok := retryAfterDelay(mk("retry-after-ms", "1500")); !ok || d != 1500*time.Millisecond {
220 t.Errorf("retry-after-ms: got %v %v", d, ok)
221 }
222 if d, ok := retryAfterDelay(mk("Retry-After", "2")); !ok || d != 2*time.Second {
223 t.Errorf("seconds: got %v %v", d, ok)
224 }
225 // retry-after-ms takes precedence over Retry-After.
226 if d, ok := retryAfterDelay(mk("retry-after-ms", "100", "Retry-After", "30")); !ok || d != 100*time.Millisecond {
227 t.Errorf("precedence: got %v %v", d, ok)
228 }
229 // HTTP-date in the past clamps to zero.
230 if d, ok := retryAfterDelay(mk("Retry-After", "Mon, 02 Jan 2006 15:04:05 GMT")); !ok || d != 0 {
231 t.Errorf("past date: got %v %v", d, ok)
232 }
233 if _, ok := retryAfterDelay(mk()); ok {
234 t.Error("empty headers should not produce a delay")
235 }
236 if _, ok := retryAfterDelay(mk("Retry-After", "soon")); ok {
237 t.Error("garbage Retry-After should not produce a delay")
238 }
239}
240
241func TestIsTerminalRateLimitBody(t *testing.T) {
242 terminal := []string{
243 `{"error":{"type":"insufficient_quota"}}`,
244 "Monthly usage limit reached",
245 "you have hit your usage limit",
246 "quota exceeded",
247 "billing hard limit",
248 }
249 for _, b := range terminal {
250 if !isTerminalRateLimitBody(b) {
251 t.Errorf("%q should be terminal", b)
252 }
253 }
254 transient := []string{
255 "Our servers are currently overloaded. Please try again later.",
256 "Rate limit reached, retry shortly",
257 "",
258 }
259 for _, b := range transient {
260 if isTerminalRateLimitBody(b) {
261 t.Errorf("%q should NOT be terminal", b)
262 }
263 }
264}
265
266func TestIsTransientConnectError(t *testing.T) {
267 good := []string{
268 "read tcp: connection reset by peer",
269 "dial tcp: connection refused",
270 "unexpected EOF",
271 "upstream connect error or disconnect/reset before headers",
272 "tls handshake error: bad cert",
273 "i/o timeout",
274 }
275 for _, m := range good {
276 if !isTransientConnectError(stringErr(m)) {
277 t.Errorf("%q should be transient", m)
278 }
279 }
280 bad := []string{
281 "context canceled",
282 "json: cannot unmarshal",
283 }
284 for _, m := range bad {
285 if isTransientConnectError(stringErr(m)) {
286 t.Errorf("%q should NOT be transient", m)
287 }
288 }
289}
290
291type stringErr string
292
293func (s stringErr) Error() string { return string(s) }