Retry, backoff and default Generate timeout #48
@@ -381,7 +381,7 @@ func (p *Provider) statusError(resp *http.Response, body []byte) error {
|
|||||||
case http.StatusTooManyRequests, statusOverloaded:
|
case http.StatusTooManyRequests, statusOverloaded:
|
||||||
return llm.NewRateLimitedError(retryAfter(resp.Header.Get("Retry-After"), time.Now()), p.apiKey)
|
return llm.NewRateLimitedError(retryAfter(resp.Header.Get("Retry-After"), time.Now()), p.apiKey)
|
||||||
case http.StatusRequestTimeout:
|
case http.StatusRequestTimeout:
|
||||||
return fmt.Errorf("HTTP %d: %w", resp.StatusCode, llm.ErrTimeout)
|
return llm.NewTimeoutStatusError(resp.StatusCode, p.apiKey)
|
||||||
default:
|
default:
|
||||||
return llm.NewProviderError(resp.StatusCode, body, p.apiKey)
|
return llm.NewProviderError(resp.StatusCode, body, p.apiKey)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -137,3 +137,32 @@ func (e *ProviderError) GoString() string { return e.text }
|
|||||||
|
|
||||||
// Is reports whether target is ErrProvider.
|
// Is reports whether target is ErrProvider.
|
||||||
func (e *ProviderError) Is(target error) bool { return target == ErrProvider }
|
func (e *ProviderError) Is(target error) bool { return target == ErrProvider }
|
||||||
|
|
||||||
|
// TimeoutStatusError is an HTTP status, such as 408, by which the vendor
|
||||||
|
// said the request timed out. It matches ErrTimeout, and Status keeps it
|
||||||
|
// apart from a transport timeout, so a caller can tell the vendor answered.
|
||||||
|
// Build it with NewTimeoutStatusError, which redacts its text.
|
||||||
|
type TimeoutStatusError struct {
|
||||||
|
status int
|
||||||
|
text string
|
||||||
|
}
|
||||||
|
|
||||||
|
// NewTimeoutStatusError returns the error for a timeout HTTP status.
|
||||||
|
// secret is the credential sent with the request, or "" when the request
|
||||||
|
// sent none; every occurrence of a non-empty secret is removed from the
|
||||||
|
// error text, and the secret itself is not kept.
|
||||||
|
func NewTimeoutStatusError(status int, secret string) *TimeoutStatusError {
|
||||||
|
text := fmt.Sprintf("HTTP %d: %s", status, ErrTimeout)
|
||||||
|
return &TimeoutStatusError{status: status, text: Redact(text, secret)}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Status returns the HTTP status code.
|
||||||
|
func (e *TimeoutStatusError) Status() int { return e.status }
|
||||||
|
|
||||||
|
func (e *TimeoutStatusError) Error() string { return e.text }
|
||||||
|
|
||||||
|
// GoString keeps %#v to the redacted Error text instead of the fields.
|
||||||
|
func (e *TimeoutStatusError) GoString() string { return e.text }
|
||||||
|
|
||||||
|
// Unwrap returns ErrTimeout.
|
||||||
|
func (e *TimeoutStatusError) Unwrap() error { return ErrTimeout }
|
||||||
|
|||||||
@@ -196,3 +196,17 @@ func TestTypedErrorsDoNotKeepSecret(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestTimeoutStatusError(t *testing.T) {
|
||||||
|
err := fmt.Errorf("openai: %w", NewTimeoutStatusError(408, "HTTP"))
|
||||||
|
if !errors.Is(err, ErrTimeout) || errors.Is(err, ErrProvider) {
|
||||||
|
t.Fatalf("%v: want ErrTimeout only", err)
|
||||||
|
}
|
||||||
|
var ts *TimeoutStatusError
|
||||||
|
if !errors.As(err, &ts) || ts.Status() != 408 {
|
||||||
|
t.Fatalf("TimeoutStatusError from %v", err)
|
||||||
|
}
|
||||||
|
if s := ts.Error() + fmt.Sprintf("%#v", ts); strings.Contains(s, "HTTP") {
|
||||||
|
t.Fatalf("text holds the secret: %s", s)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -453,7 +453,7 @@ func (p *Provider) statusError(resp *http.Response, body []byte) error {
|
|||||||
case http.StatusTooManyRequests:
|
case http.StatusTooManyRequests:
|
||||||
return llm.NewRateLimitedError(retryAfter(resp.Header.Get("Retry-After"), time.Now()), p.apiKey)
|
return llm.NewRateLimitedError(retryAfter(resp.Header.Get("Retry-After"), time.Now()), p.apiKey)
|
||||||
case http.StatusRequestTimeout:
|
case http.StatusRequestTimeout:
|
||||||
return fmt.Errorf("HTTP %d: %w", resp.StatusCode, llm.ErrTimeout)
|
return llm.NewTimeoutStatusError(resp.StatusCode, p.apiKey)
|
||||||
default:
|
default:
|
||||||
return llm.NewProviderError(resp.StatusCode, body, p.apiKey)
|
return llm.NewProviderError(resp.StatusCode, body, p.apiKey)
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -523,7 +523,8 @@ func TestConfigErrorsNeverLeakKey(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// scriptedTransport answers each request with the next scripted status and
|
// scriptedTransport answers each request with the next scripted status and
|
||||||
// body, echoing the key in every failure body, and counts the requests.
|
// body, echoing the key in every failure body, and counts the requests. A
|
||||||
|
// scripted status of 0 is a transport timeout instead of a response.
|
||||||
type scriptedTransport struct {
|
type scriptedTransport struct {
|
||||||
statuses []int
|
statuses []int
|
||||||
key string
|
key string
|
||||||
@@ -536,6 +537,9 @@ func (st *scriptedTransport) RoundTrip(r *http.Request) (*http.Response, error)
|
|||||||
}
|
}
|
||||||
status := st.statuses[st.sent]
|
status := st.statuses[st.sent]
|
||||||
st.sent++
|
st.sent++
|
||||||
|
if status == 0 {
|
||||||
|
return nil, transportTimeout{}
|
||||||
|
}
|
||||||
header := http.Header{"Content-Type": {"application/json"}}
|
header := http.Header{"Content-Type": {"application/json"}}
|
||||||
body := `{"error":{"message":"bad key ` + st.key + `"}}`
|
body := `{"error":{"message":"bad key ` + st.key + `"}}`
|
||||||
switch status {
|
switch status {
|
||||||
@@ -552,6 +556,13 @@ func (st *scriptedTransport) RoundTrip(r *http.Request) (*http.Response, error)
|
|||||||
}, nil
|
}, nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// transportTimeout is a net.Error that timed out.
|
||||||
|
type transportTimeout struct{}
|
||||||
|
|
||||||
|
func (transportTimeout) Error() string { return "i/o timeout" }
|
||||||
|
func (transportTimeout) Timeout() bool { return true }
|
||||||
|
func (transportTimeout) Temporary() bool { return true }
|
||||||
|
|
||||||
// useTransport makes the adapters send through rt for the rest of the test.
|
// useTransport makes the adapters send through rt for the rest of the test.
|
||||||
func useTransport(t *testing.T, rt http.RoundTripper) {
|
func useTransport(t *testing.T, rt http.RoundTripper) {
|
||||||
t.Helper()
|
t.Helper()
|
||||||
@@ -604,3 +615,48 @@ func TestFromSettingsRetryErrorNeverLeaksKey(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// An HTTP 408 from the vendor is a 4xx answer and is not retried, though it
|
||||||
|
// matches llm.ErrTimeout.
|
||||||
|
func TestFromSettingsDoesNotRetryHTTP408(t *testing.T) {
|
||||||
|
for _, provider := range []string{"openai", "anthropic"} {
|
||||||
|
t.Run(provider, func(t *testing.T) {
|
||||||
|
st := &scriptedTransport{statuses: []int{http.StatusRequestTimeout}, key: testKey}
|
||||||
|
useTransport(t, st)
|
||||||
|
|
||||||
|
p, err := FromSettings(llm.Settings{Provider: provider, Model: "m"}, testKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("FromSettings: %v", err)
|
||||||
|
}
|
||||||
|
_, err = p.Generate(context.Background(), plainRequest())
|
||||||
|
var ts *llm.TimeoutStatusError
|
||||||
|
if !errors.Is(err, llm.ErrTimeout) || !errors.As(err, &ts) || ts.Status() != http.StatusRequestTimeout {
|
||||||
|
t.Fatalf("Generate = %v, want an HTTP 408 ErrTimeout", err)
|
||||||
|
}
|
||||||
|
if st.sent != 1 {
|
||||||
|
t.Errorf("sent %d requests, want 1", st.sent)
|
||||||
|
}
|
||||||
|
})
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// A transport timeout carries no status and is retried.
|
||||||
|
func TestFromSettingsRetriesTransportTimeout(t *testing.T) {
|
||||||
|
st := &scriptedTransport{statuses: []int{0, http.StatusOK}, key: testKey}
|
||||||
|
useTransport(t, st)
|
||||||
|
|
||||||
|
p, err := FromSettings(llm.Settings{Provider: "openai", Model: "m"}, testKey)
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("FromSettings: %v", err)
|
||||||
|
}
|
||||||
|
resp, err := p.Generate(context.Background(), plainRequest())
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("Generate: %v", err)
|
||||||
|
}
|
||||||
|
if resp.Text != "tacos" {
|
||||||
|
t.Errorf("Text = %q, want %q", resp.Text, "tacos")
|
||||||
|
}
|
||||||
|
if st.sent != 2 {
|
||||||
|
t.Errorf("sent %d requests, want 2", st.sent)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|||||||
@@ -22,11 +22,12 @@ const (
|
|||||||
)
|
)
|
||||||
|
|
||||||
// Retrying is a Provider that retries transient Generate failures of the
|
// Retrying is a Provider that retries transient Generate failures of the
|
||||||
// Provider it wraps. It retries ErrRateLimited, ErrTimeout and a 5xx
|
// Provider it wraps. It retries ErrRateLimited, a transport ErrTimeout and
|
||||||
// ErrProvider at most MaxRetries times, and nothing else. A rate-limited
|
// a 5xx ErrProvider at most MaxRetries times, and nothing else: an error
|
||||||
// retry waits for the vendor's RetryAfter, capped at MaxRetryAfter; every
|
// carrying a 4xx HTTP status other than a rate limit, such as an HTTP 408
|
||||||
// other retry, and a rate limit that names no wait, waits a jittered
|
// TimeoutStatusError, is not retried. A rate-limited retry waits for the
|
||||||
// exponential backoff. The Request's Timeout, or DefaultTimeout, is one
|
// vendor's RetryAfter, capped at MaxRetryAfter; every other retry, and a
|
||||||
|
// rate limit that names no wait, waits a jittered exponential backoff. The Request's Timeout, or DefaultTimeout, is one
|
||||||
// deadline for the whole call, retries and waits included; a caller's
|
// deadline for the whole call, retries and waits included; a caller's
|
||||||
// earlier context deadline wins. A retry whose wait would not end before
|
// earlier context deadline wins. A retry whose wait would not end before
|
||||||
// that deadline is not made: the last error is returned at once.
|
// that deadline is not made: the last error is returned at once.
|
||||||
@@ -91,6 +92,10 @@ func (r *Retrying) Generate(ctx context.Context, req Request) (Response, error)
|
|||||||
// wait returns how long to wait before retrying after err, the retry-th
|
// wait returns how long to wait before retrying after err, the retry-th
|
||||||
// retry counting from 0, and false when err is not retried.
|
// retry counting from 0, and false when err is not retried.
|
||||||
func (r *Retrying) wait(err error, retry int) (time.Duration, bool) {
|
func (r *Retrying) wait(err error, retry int) (time.Duration, bool) {
|
||||||
|
var st interface{ Status() int }
|
||||||
|
if errors.As(err, &st) && st.Status() >= 400 && st.Status() <= 499 {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
var rl *RateLimitedError
|
var rl *RateLimitedError
|
||||||
if errors.As(err, &rl) && rl.RetryAfter > 0 {
|
if errors.As(err, &rl) && rl.RetryAfter > 0 {
|
||||||
return min(rl.RetryAfter, MaxRetryAfter), true
|
return min(rl.RetryAfter, MaxRetryAfter), true
|
||||||
|
|||||||
@@ -77,7 +77,7 @@ func TestRetryingRateLimitedWaitIsCapped(t *testing.T) {
|
|||||||
if _, err := r.Generate(context.Background(), retryReq); err != nil {
|
if _, err := r.Generate(context.Background(), retryReq); err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
if want := []time.Duration{MaxRetryAfter}; !slices.Equal(c.sleeps, want) {
|
if want := []time.Duration{20 * time.Second}; !slices.Equal(c.sleeps, want) {
|
||||||
t.Errorf("sleeps = %v, want %v", c.sleeps, want)
|
t.Errorf("sleeps = %v, want %v", c.sleeps, want)
|
||||||
}
|
}
|
||||||
assertConsumed(t, f, 2)
|
assertConsumed(t, f, 2)
|
||||||
@@ -149,11 +149,13 @@ func TestRetrying5xxThreeTimesFails(t *testing.T) {
|
|||||||
|
|
||||||
func TestRetryingDoesNotRetryPermanentErrors(t *testing.T) {
|
func TestRetryingDoesNotRetryPermanentErrors(t *testing.T) {
|
||||||
for name, err := range map[string]error{
|
for name, err := range map[string]error{
|
||||||
"401 auth": ErrAuth,
|
"401 auth": ErrAuth,
|
||||||
"400": NewProviderError(400, []byte("bad request"), ""),
|
"400": NewProviderError(400, []byte("bad request"), ""),
|
||||||
"404": NewProviderError(404, []byte("no such model"), ""),
|
"404": NewProviderError(404, []byte("no such model"), ""),
|
||||||
"bad output": ErrBadOutput,
|
"408": NewTimeoutStatusError(408, ""),
|
||||||
"other": errors.New("llm: request has no messages"),
|
"wrapped 408": fmt.Errorf("openai: generate: %w", NewTimeoutStatusError(408, "")),
|
||||||
|
"bad output": ErrBadOutput,
|
||||||
|
"other": errors.New("llm: request has no messages"),
|
||||||
} {
|
} {
|
||||||
t.Run(name, func(t *testing.T) {
|
t.Run(name, func(t *testing.T) {
|
||||||
r, f, c := newTestRetrying()
|
r, f, c := newTestRetrying()
|
||||||
@@ -170,6 +172,18 @@ func TestRetryingDoesNotRetryPermanentErrors(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// A transport timeout matches ErrTimeout like an HTTP 408, but carries no
|
||||||
|
// status, so it is retried.
|
||||||
|
func TestRetryingTransportTimeoutRetried(t *testing.T) {
|
||||||
|
r, f, _ := newTestRetrying()
|
||||||
|
f.QueueGenerate(fail(fmt.Errorf("openai: generate: %w", ErrTimeout)), ok("dinner"))
|
||||||
|
|
||||||
|
if _, err := r.Generate(context.Background(), retryReq); err != nil {
|
||||||
|
t.Fatal(err)
|
||||||
|
}
|
||||||
|
assertConsumed(t, f, 2)
|
||||||
|
}
|
||||||
|
|
||||||
func TestRetryingDeadlineShorterThanRetryAfter(t *testing.T) {
|
func TestRetryingDeadlineShorterThanRetryAfter(t *testing.T) {
|
||||||
r, f, c := newTestRetrying()
|
r, f, c := newTestRetrying()
|
||||||
rl := NewRateLimitedError(10*time.Second, "")
|
rl := NewRateLimitedError(10*time.Second, "")
|
||||||
@@ -250,7 +264,7 @@ func TestRetryingTimeoutBoundsWholeCall(t *testing.T) {
|
|||||||
timeout time.Duration
|
timeout time.Duration
|
||||||
want time.Duration
|
want time.Duration
|
||||||
}{
|
}{
|
||||||
{"default", 0, DefaultTimeout},
|
{"default", 0, 60 * time.Second},
|
||||||
{"custom", 10 * time.Second, 10 * time.Second},
|
{"custom", 10 * time.Second, 10 * time.Second},
|
||||||
} {
|
} {
|
||||||
t.Run(tc.name, func(t *testing.T) {
|
t.Run(tc.name, func(t *testing.T) {
|
||||||
|
|||||||
Reference in New Issue
Block a user