fix(proxy): harden SSE failover handling

This commit is contained in:
apale7 2026-04-19 11:24:44 +08:00
parent 136912d0ef
commit e4789d0539
4 changed files with 202 additions and 2 deletions

View File

@ -0,0 +1,26 @@
# Proxy SSE Minimal Fix
## Summary
The local Responses proxy can report `upstream_status=200` while still handing OpenCode a broken stream in two narrow cases inside `internal/proxy/server.go`:
- the upstream returns `200` with `text/event-stream` but closes before sending any bytes
- the upstream sends one SSE chunk and then stays silent longer than `streamIdleTimeout`, causing the proxy to terminate a still-valid SSE stream
This task applies the smallest possible fix in the proxy layer only, without changing CLI, desktop, app, config, or public interfaces.
## Requirements
- Keep the existing `firstByteTimeout` behavior.
- Treat `readFirstChunk()` returning `io.EOF` with zero bytes as a retryable pre-first-byte failure.
- Do not write downstream `200` for that empty first-read case.
- Preserve existing failover behavior for transport errors and `429`/`5xx` responses.
- After the first chunk succeeds, bypass `streamIdleTimeout` only for `text/event-stream` responses.
- Keep non-SSE responses on the existing idle-timeout path.
- Add regression coverage for empty-200-SSE failover and long-idle SSE continuation.
## Non-Goals
1. No proxy refactor.
2. No new config flags.
3. No changes outside `internal/proxy/server.go` and `internal/proxy/server_test.go`.

View File

@ -0,0 +1,65 @@
{
"id": "04-19-proxy-sse-minimal-fix",
"name": "04-19-proxy-sse-minimal-fix",
"title": "Fix proxy SSE minimal failure handling",
"description": "Minimal proxy fix for empty 200 SSE failover and SSE idle timeout handling",
"status": "completed",
"dev_type": "feature",
"scope": "proxy",
"package": null,
"priority": "P1",
"creator": "OpenCode",
"assignee": "OpenCode",
"createdAt": "2026-04-19",
"completedAt": "2026-04-19",
"branch": "master",
"base_branch": "master",
"worktree_path": null,
"current_phase": 6,
"next_action": [
{
"phase": 1,
"action": "brainstorm"
},
{
"phase": 2,
"action": "research"
},
{
"phase": 3,
"action": "implement"
},
{
"phase": 4,
"action": "check"
},
{
"phase": 5,
"action": "update-spec"
},
{
"phase": 6,
"action": "record-session"
}
],
"commit": null,
"pr_url": null,
"subtasks": [],
"children": [],
"parent": null,
"relatedFiles": [
".trellis/tasks/04-19-04-19-proxy-sse-minimal-fix/prd.md",
"internal/proxy/server.go",
"internal/proxy/server_test.go"
],
"notes": "Applied a minimal SSE reliability fix in the local Responses proxy. Empty `200 text/event-stream` first reads are now treated as retryable pre-first-byte failures so failover can continue before any downstream status is written. After the first SSE chunk arrives, the proxy no longer applies `streamIdleTimeout` to later SSE reads, while non-SSE responses keep the existing timeout behavior. Added regression coverage for empty-200 SSE failover and long-idle SSE continuation, and verified with `go test ./internal/proxy`.",
"meta": {
"tests": [
"go test ./internal/proxy"
],
"regressions": [
"empty 200 SSE failover",
"SSE bypasses idle timeout after first chunk"
]
}
}

View File

@ -8,6 +8,7 @@ import (
"fmt" "fmt"
"io" "io"
"log" "log"
"mime"
"net" "net"
"net/http" "net/http"
"strings" "strings"
@ -296,13 +297,18 @@ func (s *Server) tryOnce(
} }
if errors.Is(firstErr, io.EOF) { if errors.Is(firstErr, io.EOF) {
if len(firstChunk) == 0 { if len(firstChunk) == 0 {
firstChunk = nil return false, true, fmt.Errorf("upstream closed before first byte"), nil
} }
} else { } else {
return false, true, fmt.Errorf("upstream first read: %w", firstErr), nil return false, true, fmt.Errorf("upstream first read: %w", firstErr), nil
} }
} }
isEventStream := false
if mediaType, _, parseErr := mime.ParseMediaType(resp.Header.Get("Content-Type")); parseErr == nil {
isEventStream = mediaType == "text/event-stream"
}
s.logger.Printf("alias=%s attempt=%d provider=%s remote_model=%s upstream_status=%d", aliasName, attempt, provider.ID, target.Model, resp.StatusCode) s.logger.Printf("alias=%s attempt=%d provider=%s remote_model=%s upstream_status=%d", aliasName, attempt, provider.ID, target.Model, resp.StatusCode)
s.writeDebugHeaders(w, aliasName, provider.ID, target.Model, attempt, failoverCount) s.writeDebugHeaders(w, aliasName, provider.ID, target.Model, attempt, failoverCount)
copyResponseHeaders(w.Header(), resp.Header) copyResponseHeaders(w.Header(), resp.Header)
@ -318,7 +324,15 @@ func (s *Server) tryOnce(
} }
buf := make([]byte, 16<<10) buf := make([]byte, 16<<10)
for { for {
n, rerr := readChunkWithTimeout(resp.Body, buf, streamIdleTimeout) var (
n int
rerr error
)
if isEventStream {
n, rerr = resp.Body.Read(buf)
} else {
n, rerr = readChunkWithTimeout(resp.Body, buf, streamIdleTimeout)
}
if n > 0 { if n > 0 {
if _, werr := w.Write(buf[:n]); werr != nil { if _, werr := w.Write(buf[:n]); werr != nil {
return true, false, werr, nil return true, false, werr, nil
@ -331,6 +345,7 @@ func (s *Server) tryOnce(
if errors.Is(rerr, io.EOF) { if errors.Is(rerr, io.EOF) {
return true, false, nil, nil return true, false, nil, nil
} }
s.logger.Printf("alias=%s attempt=%d provider=%s remote_model=%s upstream body read failed after response start: %v", aliasName, attempt, provider.ID, target.Model, rerr)
return true, false, rerr, nil return true, false, rerr, nil
} }
} }

View File

@ -107,6 +107,100 @@ func TestHandleResponsesFailsOverOn429(t *testing.T) {
} }
} }
func TestHandleResponsesFailsOverOnEmptySSE200(t *testing.T) {
t.Parallel()
first := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
}))
defer first.Close()
second := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("data: ok\n\n"))
}))
defer second.Close()
srv := New(&config.Config{
Server: config.Server{APIKey: config.DefaultLocalAPIKey},
Providers: []config.Provider{
{ID: "p1", BaseURL: first.URL + "/v1"},
{ID: "p2", BaseURL: second.URL + "/v1"},
},
Aliases: []config.Alias{{
Alias: "gpt-5.4",
Enabled: true,
Targets: []config.Target{{Provider: "p1", Model: "up-1", Enabled: true}, {Provider: "p2", Model: "up-2", Enabled: true}},
}},
})
req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","stream":true}`))
req.Header.Set("Authorization", "Bearer "+config.DefaultLocalAPIKey)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "text/event-stream")
rr := httptest.NewRecorder()
srv.handleResponses(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", rr.Code, http.StatusOK)
}
if body := rr.Body.String(); body != "data: ok\n\n" {
t.Fatalf("body = %q, want SSE payload from second upstream", body)
}
if got := rr.Header().Get("X-OCSWITCH-Attempt"); got != "2" {
t.Fatalf("X-OCSWITCH-Attempt = %q, want 2", got)
}
}
func TestHandleResponsesSSEBypassesIdleTimeout(t *testing.T) {
oldTimeout := streamIdleTimeout
streamIdleTimeout = 30 * time.Millisecond
defer func() { streamIdleTimeout = oldTimeout }()
upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "text/event-stream")
w.WriteHeader(http.StatusOK)
_, _ = w.Write([]byte("data: first\n\n"))
if flusher, ok := w.(http.Flusher); ok {
flusher.Flush()
}
time.Sleep(90 * time.Millisecond)
_, _ = w.Write([]byte("data: second\n\n"))
if flusher, ok := w.(http.Flusher); ok {
flusher.Flush()
}
}))
defer upstream.Close()
srv := New(&config.Config{
Server: config.Server{APIKey: config.DefaultLocalAPIKey},
Providers: []config.Provider{{ID: "p1", BaseURL: upstream.URL + "/v1"}},
Aliases: []config.Alias{{
Alias: "gpt-5.4",
Enabled: true,
Targets: []config.Target{{Provider: "p1", Model: "up-1", Enabled: true}},
}},
})
req := httptest.NewRequest(http.MethodPost, "/v1/responses", strings.NewReader(`{"model":"gpt-5.4","stream":true}`))
req.Header.Set("Authorization", "Bearer "+config.DefaultLocalAPIKey)
req.Header.Set("Content-Type", "application/json")
req.Header.Set("Accept", "text/event-stream")
rr := httptest.NewRecorder()
srv.handleResponses(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status = %d, want %d", rr.Code, http.StatusOK)
}
if body := rr.Body.String(); body != "data: first\n\ndata: second\n\n" {
t.Fatalf("body = %q, want both SSE chunks", body)
}
}
func TestHandleResponsesDoesNotFailOverOn400(t *testing.T) { func TestHandleResponsesDoesNotFailOverOn400(t *testing.T) {
t.Parallel() t.Parallel()