diff --git a/classify.go b/classify.go index 51c3668..9df8cc7 100644 --- a/classify.go +++ b/classify.go @@ -224,7 +224,11 @@ func classifySSEError(data string) *failure { return &failure{transient: true, status: 429, atype: "rate_limit_error", code: "sse_rate_limit", message: e.Error.Message} case "api_error", "timeout_error", "": return &failure{transient: true, status: 502, atype: "api_error", code: "sse_api_error", message: e.Error.Message} - case "invalid_request_error": // request-shape: can never succeed, surface it + case "invalid_request_error": + // Only this verified transport failure overrides the request-error label. + if e.Error.Message == "stream error: stream disconnected before completion: stream closed before response.completed" { + return &failure{transient: true, status: 502, atype: "api_error", code: "sse_stream_disconnect", message: e.Error.Message} + } return &failure{transient: false, status: 400, atype: e.Error.Type, code: "sse_request_shape", message: e.Error.Message} default: // authentication_error, permission_error, not_found_error, unknown -> retry return &failure{transient: true, status: 502, atype: "api_error", code: "sse_retryable", message: e.Error.Message} diff --git a/docs/TROUBLESHOOTING.md b/docs/TROUBLESHOOTING.md index 1600b4e..6cf945c 100644 --- a/docs/TROUBLESHOOTING.md +++ b/docs/TROUBLESHOOTING.md @@ -76,6 +76,18 @@ specific `code` is `request_timeout`, `rate_limit_exceeded`, or example, `request_timeout` with “stream disconnected before completion” remains retryable instead of becoming an invalid-request failure. +For Anthropic SSE `event: error`, the proxy also recognizes one verified transport +disconnect mislabeled `invalid_request_error`. The complete message must be exactly: + +```text +stream error: stream disconnected before completion: stream closed before response.completed +``` + +This failure is classified as `sse_stream_disconnect`. Before commit, failed +output is discarded and the existing local and client retry budgets apply, within +the current deadline and buffering window. After commit, the stream ends without +transparent replay. This exception does not change retry limits or timeouts. + If retries continue, check the gateway's credentials, account state, availability, and rate limits. A permanently bad key or exhausted balance will not improve without intervention, even though the proxy allows time for recovery. Generic @@ -108,9 +120,11 @@ are returned with `X-Should-Retry: false`; terminal Responses errors use HTTP Anthropic HTTP errors use recognized message signatures. A bare HTTP `invalid_request_error` without one can still retry; an Anthropic in-stream -`invalid_request_error` is terminal. Responses classification uses error codes, -types, and recognized messages, with known transient codes taking precedence over -generic invalid-request types. Gateway retry headers override HTTP heuristics. +`invalid_request_error` is terminal except for the exact transport-disconnect +signature above. Invalid schemas, malformed tool messages, and context-limit +failures remain terminal. Responses classification uses error codes, types, and +recognized messages, with known transient codes taking precedence over generic +invalid-request types. Gateway retry headers override HTTP heuristics. ### Codex context compaction diff --git a/e2e_test.go b/e2e_test.go index a8af43b..7dd7c05 100644 --- a/e2e_test.go +++ b/e2e_test.go @@ -1,6 +1,7 @@ package main import ( + "bufio" "bytes" "io" "log" @@ -395,6 +396,125 @@ func TestE2ETransactionalLocalRetryTruncateThenSuccess(t *testing.T) { } } +func TestE2ESSEDisconnectRetryBudget(t *testing.T) { + for _, tc := range []struct { + name string + retries int + recover bool + backoff, window, deadline time.Duration + attempts, status int + }{ + {name: "recovered", retries: 1, recover: true, attempts: 2, status: 200}, + {name: "local_disabled", recover: true, attempts: 1, status: 503}, + {name: "local_exhausted", retries: 1, attempts: 2, status: 503}, + {name: "deadline_insufficient", retries: 1, recover: true, backoff: time.Second, deadline: time.Second, attempts: 1, status: 503}, + {name: "window_insufficient", retries: 1, recover: true, backoff: time.Second, window: 500 * time.Millisecond, attempts: 1, status: 503}, + } { + t.Run(tc.name, func(t *testing.T) { + var hits atomic.Int32 + prefix := strings.Replace(goodStream[:strings.Index(goodStream, "event: content_block_stop")], "Hi", "discarded-attempt", 1) + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + if hits.Add(1) > 1 && tc.recover { + io.WriteString(w, goodStream) + return + } + io.WriteString(w, prefix+mislabeledDisconnectStream) + })) + defer up.Close() + setupForTest(up.URL) + t.Cleanup(func() { cfg = testConfig() }) + cfg.txLocalRetries = tc.retries + cfg.localBackoffCap = tc.backoff + if tc.window > 0 { + cfg.keepaliveMs = tc.window + } + if tc.deadline > 0 { + cfg.maxRequestDur = tc.deadline + } + + rec := doStream(`{"stream":true,"model":"m"}`) + body := rec.Body.String() + if rec.Code != tc.status || int(hits.Load()) != tc.attempts { + t.Fatalf("want status=%d attempts=%d, got status=%d attempts=%d body=%s", tc.status, tc.attempts, rec.Code, hits.Load(), body) + } + if strings.Contains(body, "discarded-attempt") { + t.Fatalf("failed attempt leaked output: %s", body) + } + if tc.status == 200 { + if rec.Header().Get("X-Steady-Proxy-Mode") != "buffered" || !strings.Contains(body, "message_stop") || !strings.Contains(body, "Hi") || strings.Contains(body, "event: error") { + t.Fatalf("recovery did not deliver a complete buffered response: headers=%v body=%s", rec.Header(), body) + } + } else if rec.Header().Get("X-Should-Retry") != "true" || rec.Header().Get("X-Steady-Proxy-Reason") != "sse_stream_disconnect" || !strings.Contains(body, `"type":"api_error"`) { + t.Fatalf("uncommitted disconnect must retain the client retry signal: headers=%v body=%s", rec.Header(), body) + } + }) + } +} + +func TestE2ESSEDisconnectAfterCommitDoesNotReplay(t *testing.T) { + var hits atomic.Int32 + releaseError := make(chan struct{}, 1) + prefix := strings.Replace(goodStream[:strings.Index(goodStream, "event: content_block_stop")], "Hi", "committed-attempt", 1) + up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + if hits.Add(1) > 1 { + io.WriteString(w, goodStream) + return + } + io.WriteString(w, prefix) + w.(http.Flusher).Flush() + select { + case <-releaseError: + io.WriteString(w, mislabeledDisconnectStream) + case <-r.Context().Done(): + } + })) + defer up.Close() + setupForTest(up.URL) + t.Cleanup(func() { cfg = testConfig() }) + cfg.keepaliveMs = 20 * time.Millisecond + cfg.txLocalRetries = 2 + cfg.localBackoffCap = 0 + proxy := httptest.NewServer(http.HandlerFunc(handle)) + defer proxy.Close() + downstream := &http.Client{Timeout: 3 * time.Second} + resp, err := downstream.Post(proxy.URL+"/v1/messages", "application/json", strings.NewReader(`{"stream":true,"model":"m"}`)) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + if resp.StatusCode != 200 || resp.Header.Get("X-Steady-Proxy-Mode") != "live" { + t.Fatalf("want committed live response, got status=%d headers=%v", resp.StatusCode, resp.Header) + } + reader := bufio.NewReader(resp.Body) + var received strings.Builder + for { + line, err := reader.ReadString('\n') + if err != nil { + t.Fatalf("reading committed content: %v", err) + } + received.WriteString(line) + if strings.Contains(line, "committed-attempt") { + break + } + } + // Release the error only after the client has observed actual response content. + releaseError <- struct{}{} + rest, err := io.ReadAll(reader) + if err != nil { + t.Fatal(err) + } + received.Write(rest) + body := received.String() + if hits.Load() != 1 { + t.Fatalf("committed failure replayed upstream: attempts=%d body=%s", hits.Load(), body) + } + if strings.Contains(body, "message_stop") || strings.Contains(body, "event: error") || strings.Contains(body, streamDisconnectMessage) || strings.Contains(body, `"text":"Hi"`) { + t.Fatalf("committed failure must end the partial stream without an error or replacement response: %s", body) + } +} + func TestE2ETransactionalLocalRetryDoesNotRetryPermanentShape(t *testing.T) { var hits atomic.Int32 up := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { diff --git a/sse_test.go b/sse_test.go index eb45fda..dfe1bf1 100644 --- a/sse_test.go +++ b/sse_test.go @@ -31,6 +31,10 @@ data: {"type":"message_stop"} ` +const streamDisconnectMessage = "stream error: stream disconnected before completion: stream closed before response.completed" + +const mislabeledDisconnectStream = "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"" + streamDisconnectMessage + "\"}}\n\n" + func init() { cfg = testConfig() } func capture(t *testing.T, s string) (*httptest.ResponseRecorder, *failure) { @@ -482,11 +486,46 @@ func TestCaptureSSEErrorOverloaded(t *testing.T) { } } -func TestCaptureSSEErrorPermanent(t *testing.T) { - s := "event: error\ndata: {\"type\":\"error\",\"error\":{\"type\":\"invalid_request_error\",\"message\":\"bad\"}}\n\n" - _, f := capture(t, s) - if f == nil || f.transient { - t.Fatalf("want permanent failure, got %+v", f) +func TestCaptureSSEErrorClassification(t *testing.T) { + useDefaultConfig(t) + for _, tc := range []struct { + name, typ, message string + transient bool + }{ + {"mislabeled_disconnect", "invalid_request_error", streamDisconnectMessage, true}, + {"typed_disconnect", "timeout_error", streamDisconnectMessage, true}, + {"unknown_request_error", "invalid_request_error", "bad", false}, + {"invalid_schema", "invalid_request_error", "invalid tool schema", false}, + {"malformed_tool_message", "invalid_request_error", "messages.0: unexpected tool_use_id in tool_result", false}, + {"context_limit", "invalid_request_error", "prompt is too long for context", false}, + {"stream_validation", "invalid_request_error", "invalid tool schema for stream", false}, + {"stream_error_validation", "invalid_request_error", "stream error: invalid tool schema", false}, + {"partial_disconnect", "invalid_request_error", "stream disconnected before completion", false}, + {"different_stream_error", "invalid_request_error", "stream error: stream disconnected before completion: invalid tool schema", false}, + {"disconnect_with_validation", "invalid_request_error", streamDisconnectMessage + ": invalid tool schema", false}, + } { + t.Run(tc.name, func(t *testing.T) { + data, err := json.Marshal(map[string]any{ + "type": "error", "error": map[string]string{"type": tc.typ, "message": tc.message}, + }) + if err != nil { + t.Fatal(err) + } + rec, f := capture(t, "event: error\ndata: "+string(data)+"\n\n") + if f == nil || f.transient != tc.transient || f.message != tc.message { + t.Fatalf("want transient=%v with original message, got %+v", tc.transient, f) + } + status, typ := http.StatusBadRequest, "invalid_request_error" + if tc.transient { + status, typ = http.StatusBadGateway, "api_error" + } + if f.status != status || f.atype != typ { + t.Fatalf("want status=%d type=%s, got %+v", status, typ, f) + } + if rec.Body.Len() != 0 { + t.Fatalf("failed attempt leaked output: %s", rec.Body.String()) + } + }) } }