From dcdb6559b885159f17373f7e4af040b008e71e68 Mon Sep 17 00:00:00 2001 From: caigee-cmd <210803857+caigee-cmd@users.noreply.github.com> Date: Thu, 22 Oct 2026 02:19:56 +0800 Subject: [PATCH 1/2] fix(qoder): bound ensureMeaningfulUpstream after 200 (#344) Reuse QODER_STREAM_TTFB_MS / Direct.ttfb so heartbeat-only SSE after a 200 fails over with upstream_empty instead of hanging forever. Apply the same deadline to qwenwork for consistency. --- .../qoder-pat-meaningful-upstream-timeout.md | 9 ++ deploy/README.md | 2 +- internal/providers/qoder/direct.go | 46 ++++++- internal/providers/qoder/direct_test.go | 130 ++++++++++++++++++ internal/providers/qwenwork/client.go | 2 +- internal/providers/qwenwork/sse.go | 38 ++++- .../qwenwork/sse_meaningful_timeout_test.go | 63 +++++++++ 7 files changed, 281 insertions(+), 9 deletions(-) create mode 100644 changelog/unreleased/qoder-pat-meaningful-upstream-timeout.md create mode 100644 internal/providers/qwenwork/sse_meaningful_timeout_test.go diff --git a/changelog/unreleased/qoder-pat-meaningful-upstream-timeout.md b/changelog/unreleased/qoder-pat-meaningful-upstream-timeout.md new file mode 100644 index 00000000..f0c65c82 --- /dev/null +++ b/changelog/unreleased/qoder-pat-meaningful-upstream-timeout.md @@ -0,0 +1,9 @@ +### English + +- Qoder PAT in-process ChatStream now bounds the wait for a meaningful SSE data line after a 200 (reuses `QODER_STREAM_TTFB_MS`); heartbeat-only upstreams fail over with `upstream_empty` instead of hanging. (#344) +- QwenWork ChatStream applies the same meaningful-upstream deadline for consistency. + +### 中文 + +- Qoder PAT 进程内 ChatStream 在收到 200 后等待有意义 SSE 数据行时也会受 `QODER_STREAM_TTFB_MS` 约束；仅心跳的上游会以 `upstream_empty` 失败并切换，而不再挂起。（#344） +- QwenWork ChatStream 同样加上有意义上游数据的等待上限，保持一致。 diff --git a/deploy/README.md b/deploy/README.md index ed8c74b9..54df352c 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -134,7 +134,7 @@ console; environment variables cannot replace it. | `QODER_MAX_RETRY_ACCOUNTS` | `4` | Maximum accounts attempted for one request (1-64) | | `QODER_SSE_DIAGNOSTIC_MODELS` | empty | Comma-separated Qoder model IDs for redacted SSE diagnostics in Runtime Logs; `*` enables all | | `QODER_WORKER_BASE_PORT` | `32100` | Internal child-runtime port range | -| `QODER_STREAM_TTFB_MS` | `30000` | Qoder stream first-byte (response head) deadline in ms, Go side and worker; min `5000`; timeout fails over to the next account | +| `QODER_STREAM_TTFB_MS` | `30000` | Qoder stream first-byte deadline in ms (response head + first meaningful SSE data line after 200), Go side and worker; min `5000`; timeout fails over to the next account | | `QODER_STREAM_TOTAL_MS` | `300000` | Qoder worker mid-stream stall timeout in ms: aborts only after this long with no new chunk (resets per chunk, not a wall-clock cap); never below the TTFB value | | `QODER_SIGNATURE_SECRET` | built-in | Override the COSY request-signing secret for in-process Qoder; leave unset unless upstream rotates it | | `QODER_PROXY_URL` | empty | Initial global outbound proxy: `http(s)://`, `direct`, or `none`; saved console settings take precedence | diff --git a/internal/providers/qoder/direct.go b/internal/providers/qoder/direct.go index bdef0809..c38176b6 100644 --- a/internal/providers/qoder/direct.go +++ b/internal/providers/qoder/direct.go @@ -12,6 +12,7 @@ import ( "sort" "strings" "sync" + "sync/atomic" "time" "github.com/caigee-cmd/cli2api/internal/accounts" @@ -29,8 +30,9 @@ type Direct struct { bridgeOf func(ctx context.Context, accountID string) (*Bridge, error) http *http.Client now func() time.Time - // ttfb bounds how long ChatStream waits for the upstream response head. - // Zero falls back to defaultFirstByteTimeout inside doWithFirstByteTimeout. + // ttfb bounds how long ChatStream waits for the upstream response head + // and, after a 200, for the first meaningful SSE data line + // (ensureMeaningfulUpstream). Zero falls back to defaultFirstByteTimeout. ttfb time.Duration } @@ -205,7 +207,7 @@ func (d *Direct) ChatStream(ctx context.Context, accountID string, req translate resp.Body.Close() return nil, providers.ResolvedChat{}, upstreamStatusError(resp.StatusCode, raw) } - guarded, err := ensureMeaningfulUpstream(resp) + guarded, err := ensureMeaningfulUpstream(resp, d.ttfb) if err != nil { // ensureMeaningfulUpstream closes the body on failure; no client bytes // have been written yet, so the executor can still fail over. @@ -251,7 +253,32 @@ func emptyUpstreamError() error { // completion. Peek before the first client byte so the failure can still be // classified and retried on another account. On success the buffered bytes // are replayed ahead of the remaining body, preserving real-time streaming. -func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) { +// +// timeout bounds how long to wait for a meaningful data line after the +// response head arrives. Zero falls back to defaultFirstByteTimeout so the +// bound matches QODER_STREAM_TTFB_MS / Direct.ttfb. A stuck heartbeat-only +// upstream closes the body and returns emptyUpstreamError so the executor +// can fail over instead of hanging forever (doWithFirstByteTimeout only +// covers headers). +func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http.Response, error) { + if resp == nil || resp.Body == nil { + return nil, emptyUpstreamError() + } + if timeout <= 0 { + timeout = defaultFirstByteTimeout + } + + // Close the body when the deadline fires so a blocked ReadString unblocks. + // Success stops the timer before returning the live peeked body; if the + // closer already won the race, treat it as emptyUpstreamError rather than + // handing the caller a half-closed stream. + var expired atomic.Bool + timer := time.AfterFunc(timeout, func() { + expired.Store(true) + _ = resp.Body.Close() + }) + stopTimer := func() { _ = timer.Stop() } + br := bufio.NewReader(resp.Body) var pending bytes.Buffer for { @@ -259,12 +286,21 @@ func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) { if line != "" { pending.WriteString(line) if data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:"); ok && meaningfulDataLine(strings.TrimSpace(data)) { + stopTimer() + if expired.Load() { + _ = resp.Body.Close() + return nil, emptyUpstreamError() + } resp.Body = peekedBody{Reader: io.MultiReader(&pending, br), Closer: resp.Body} return resp, nil } } if readErr != nil { - resp.Body.Close() + stopTimer() + _ = resp.Body.Close() + if expired.Load() { + return nil, emptyUpstreamError() + } if errors.Is(readErr, io.EOF) { return nil, emptyUpstreamError() } diff --git a/internal/providers/qoder/direct_test.go b/internal/providers/qoder/direct_test.go index 1e61b44a..2304e003 100644 --- a/internal/providers/qoder/direct_test.go +++ b/internal/providers/qoder/direct_test.go @@ -245,3 +245,133 @@ func TestDirectChatStreamSalvagesTruncatedStreamWithCachedUsage(t *testing.T) { t.Fatalf("missing cached_tokens in terminal chunk: %s", text) } } + +// Heartbeat-only upstream that never becomes meaningful must time out with +// upstream_empty so the executor can fail over. Without the bound, +// ensureMeaningfulUpstream would hang forever after a 200 (issue #344). +func TestDirectChatStreamTimesOutHeartbeatOnlyUpstream(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + heartbeat := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n" + for { + select { + case <-r.Context().Done(): + return + default: + _, _ = io.WriteString(w, heartbeat) + if flusher != nil { + flusher.Flush() + } + time.Sleep(20 * time.Millisecond) + } + } + })) + defer upstream.Close() + direct := newDirectFor(t, upstream) + direct.ttfb = 80 * time.Millisecond + + started := time.Now() + resp, _, err := direct.ChatStream(context.Background(), "acc", translate.ChatRequest{Model: "auto"}) + elapsed := time.Since(started) + if err == nil { + resp.Body.Close() + t.Fatal("expected timeout error for heartbeat-only upstream") + } + var providerErr *providers.Error + if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" { + t.Fatalf("err = %#v", err) + } + if resp != nil { + t.Fatalf("resp = %v, want nil", resp) + } + // Must fail within the bound (plus small slack), not hang for minutes. + if elapsed > 2*time.Second { + t.Fatalf("took %s, want bound near ttfb", elapsed) + } +} + +// A meaningful data line that arrives within the deadline must succeed, and +// the peeked heartbeat + content bytes must still be replayed to the client. +func TestDirectChatStreamPreservesBytesAfterHeartbeatThenContent(t *testing.T) { + upstream := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/event-stream") + w.WriteHeader(http.StatusOK) + flusher, _ := w.(http.Flusher) + _, _ = io.WriteString(w, "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n") + if flusher != nil { + flusher.Flush() + } + time.Sleep(30 * time.Millisecond) + _, _ = io.WriteString(w, "data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\"OK\"},\"finish_reason\":\"stop\"}]}}\n\n") + if flusher != nil { + flusher.Flush() + } + })) + defer upstream.Close() + direct := newDirectFor(t, upstream) + direct.ttfb = 500 * time.Millisecond + + resp, _, err := direct.ChatStream(context.Background(), "acc", translate.ChatRequest{Model: "auto"}) + if err != nil { + t.Fatal(err) + } + defer resp.Body.Close() + body, _ := io.ReadAll(resp.Body) + text := string(body) + if !strings.Contains(text, `"content":"OK"`) { + t.Fatalf("meaningful content lost: %s", text) + } + if !strings.Contains(text, `"finish_reason":"stop"`) { + t.Fatalf("finish_reason lost: %s", text) + } +} + +func TestEnsureMeaningfulUpstreamTimesOutHeartbeatOnly(t *testing.T) { + pr, pw := io.Pipe() + go func() { + defer pw.Close() + heartbeat := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n" + for i := 0; i < 100; i++ { + if _, err := io.WriteString(pw, heartbeat); err != nil { + return + } + time.Sleep(15 * time.Millisecond) + } + }() + resp := &http.Response{StatusCode: http.StatusOK, Body: pr} + started := time.Now() + out, err := ensureMeaningfulUpstream(resp, 60*time.Millisecond) + elapsed := time.Since(started) + if err == nil { + out.Body.Close() + t.Fatal("expected timeout") + } + var providerErr *providers.Error + if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" { + t.Fatalf("err = %#v", err) + } + if out != nil { + t.Fatalf("out = %v, want nil", out) + } + if elapsed > 2*time.Second { + t.Fatalf("took %s", elapsed) + } +} + +func TestEnsureMeaningfulUpstreamPreservesBufferedBytes(t *testing.T) { + body := "data: {\"body\":{\"choices\":[{\"delta\":{},\"finish_reason\":null}]}}\n\n" + + "data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\"hi\"},\"finish_reason\":\"stop\"}]}}\n\n" + + "data: {\"body\":{\"choices\":[{\"delta\":{\"content\":\" more\"}}]}}\n\n" + resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body))} + out, err := ensureMeaningfulUpstream(resp, time.Second) + if err != nil { + t.Fatal(err) + } + defer out.Body.Close() + got, _ := io.ReadAll(out.Body) + if string(got) != body { + t.Fatalf("peeked replay mismatch:\n got %q\nwant %q", got, body) + } +} diff --git a/internal/providers/qwenwork/client.go b/internal/providers/qwenwork/client.go index 36b78448..edc200d9 100644 --- a/internal/providers/qwenwork/client.go +++ b/internal/providers/qwenwork/client.go @@ -680,7 +680,7 @@ func (c *Client) ChatStream(ctx context.Context, accountID string, req translate resp.Body.Close() return nil, providers.ResolvedChat{}, classifiedError(resp.StatusCode, raw) } - guarded, err := ensureMeaningfulUpstream(resp) + guarded, err := ensureMeaningfulUpstream(resp, defaultMeaningfulUpstreamTimeout) if err != nil { // ensureMeaningfulUpstream closes the body on failure; no client // bytes have been written yet, so the executor can still fail over. diff --git a/internal/providers/qwenwork/sse.go b/internal/providers/qwenwork/sse.go index d32f3fde..d33d515b 100644 --- a/internal/providers/qwenwork/sse.go +++ b/internal/providers/qwenwork/sse.go @@ -11,6 +11,7 @@ import ( "sort" "strconv" "strings" + "sync/atomic" "time" "github.com/caigee-cmd/cli2api/internal/accounts" @@ -590,11 +591,35 @@ func emptyUpstreamError() *providers.Error { } } +// defaultMeaningfulUpstreamTimeout bounds how long ChatStream waits after a +// 200 for a meaningful SSE data line. Matches qoder's defaultFirstByteTimeout +// so heartbeat-only upstreams fail over instead of hanging the client. +const defaultMeaningfulUpstreamTimeout = 30 * time.Second + // ensureMeaningfulUpstream peeks the upstream SSE body before anything is // relayed to the client, so an empty or error-only upstream can still fail // over to another account. On success the buffered bytes are replayed ahead // of the remaining body, preserving real-time streaming. -func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) { +// +// timeout bounds the wait for a meaningful data line after the response head. +// Zero falls back to defaultMeaningfulUpstreamTimeout. A stuck heartbeat-only +// upstream closes the body and returns emptyUpstreamError so the executor can +// fail over (same hang shape as qoder PAT in-process ChatStream). +func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http.Response, error) { + if resp == nil || resp.Body == nil { + return nil, emptyUpstreamError() + } + if timeout <= 0 { + timeout = defaultMeaningfulUpstreamTimeout + } + + var expired atomic.Bool + timer := time.AfterFunc(timeout, func() { + expired.Store(true) + _ = resp.Body.Close() + }) + stopTimer := func() { _ = timer.Stop() } + br := bufio.NewReader(resp.Body) var pending bytes.Buffer for { @@ -603,6 +628,11 @@ func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) { pending.WriteString(line) if data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:"); ok { if frame, meaningful := meaningfulDataLine(strings.TrimSpace(data)); meaningful { + stopTimer() + if expired.Load() { + _ = resp.Body.Close() + return nil, emptyUpstreamError() + } if frame.errorOnly() { // Nothing has reached the client yet: return the // classified error so the executor cools this account @@ -616,7 +646,11 @@ func ensureMeaningfulUpstream(resp *http.Response) (*http.Response, error) { } } if readErr != nil { - resp.Body.Close() + stopTimer() + _ = resp.Body.Close() + if expired.Load() { + return nil, emptyUpstreamError() + } if errors.Is(readErr, io.EOF) { return nil, emptyUpstreamError() } diff --git a/internal/providers/qwenwork/sse_meaningful_timeout_test.go b/internal/providers/qwenwork/sse_meaningful_timeout_test.go new file mode 100644 index 00000000..795e9103 --- /dev/null +++ b/internal/providers/qwenwork/sse_meaningful_timeout_test.go @@ -0,0 +1,63 @@ +package qwenwork + +import ( + "errors" + "io" + "net/http" + "strings" + "testing" + "time" + + "github.com/caigee-cmd/cli2api/internal/providers" +) + +// Heartbeat-only forever stream must time out with upstream_empty (same hang +// shape as qoder PAT in-process ChatStream; issue #344). +func TestEnsureMeaningfulUpstreamTimesOutHeartbeatOnly(t *testing.T) { + pr, pw := io.Pipe() + go func() { + defer pw.Close() + // Empty body choices are not meaningful for qwenwork either. + heartbeat := "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{},\\\"finish_reason\\\":null}]}\",\"statusCodeValue\":200}\n\n" + for i := 0; i < 100; i++ { + if _, err := io.WriteString(pw, heartbeat); err != nil { + return + } + time.Sleep(15 * time.Millisecond) + } + }() + resp := &http.Response{StatusCode: http.StatusOK, Body: pr} + started := time.Now() + out, err := ensureMeaningfulUpstream(resp, 60*time.Millisecond) + elapsed := time.Since(started) + if err == nil { + out.Body.Close() + t.Fatal("expected timeout") + } + var providerErr *providers.Error + if !errors.As(err, &providerErr) || providerErr.Code != "upstream_empty" { + t.Fatalf("err = %#v", err) + } + if out != nil { + t.Fatalf("out = %v, want nil", out) + } + if elapsed > 2*time.Second { + t.Fatalf("took %s", elapsed) + } +} + +func TestEnsureMeaningfulUpstreamPreservesBufferedBytes(t *testing.T) { + body := "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{},\\\"finish_reason\\\":null}]}\",\"statusCodeValue\":200}\n\n" + + "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{\\\"content\\\":\\\"hi\\\"},\\\"finish_reason\\\":\\\"stop\\\"}]}\",\"statusCodeValue\":200}\n\n" + + "data: {\"headers\":{},\"body\":\"{\\\"choices\\\":[{\\\"delta\\\":{\\\"content\\\":\\\" more\\\"}}]}\",\"statusCodeValue\":200}\n\n" + resp := &http.Response{StatusCode: http.StatusOK, Body: io.NopCloser(strings.NewReader(body))} + out, err := ensureMeaningfulUpstream(resp, time.Second) + if err != nil { + t.Fatal(err) + } + defer out.Body.Close() + got, _ := io.ReadAll(out.Body) + if string(got) != body { + t.Fatalf("peeked replay mismatch:\n got %q\nwant %q", got, body) + } +} From 31389649af9dad2ca5f610bd84bde70da97454d3 Mon Sep 17 00:00:00 2001 From: caigee-cmd <210803857+caigee-cmd@users.noreply.github.com> Date: Thu, 22 Oct 2026 02:24:22 +0800 Subject: [PATCH 2/2] fix(qoder): close meaningful-upstream body at most once The TTFB timeout timer and the read-error path both Close the body; cancelOnCloseBody.Close is not concurrency-safe, so race detector failed TestDirectChatStreamTimesOutHeartbeatOnlyUpstream. Use sync.Once. --- internal/providers/qoder/direct.go | 14 ++++++++++---- internal/providers/qwenwork/sse.go | 14 ++++++++++---- 2 files changed, 20 insertions(+), 8 deletions(-) diff --git a/internal/providers/qoder/direct.go b/internal/providers/qoder/direct.go index c38176b6..a7bb9653 100644 --- a/internal/providers/qoder/direct.go +++ b/internal/providers/qoder/direct.go @@ -268,14 +268,20 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http timeout = defaultFirstByteTimeout } - // Close the body when the deadline fires so a blocked ReadString unblocks. + // Close at most once: the timeout timer and the read-error path both need + // to finish the body, but cancelOnCloseBody.Close is not concurrency-safe. // Success stops the timer before returning the live peeked body; if the // closer already won the race, treat it as emptyUpstreamError rather than // handing the caller a half-closed stream. + var closeOnce sync.Once + closeBody := func() { + closeOnce.Do(func() { _ = resp.Body.Close() }) + } + var expired atomic.Bool timer := time.AfterFunc(timeout, func() { expired.Store(true) - _ = resp.Body.Close() + closeBody() }) stopTimer := func() { _ = timer.Stop() } @@ -288,7 +294,7 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http if data, ok := strings.CutPrefix(strings.TrimSpace(line), "data:"); ok && meaningfulDataLine(strings.TrimSpace(data)) { stopTimer() if expired.Load() { - _ = resp.Body.Close() + closeBody() return nil, emptyUpstreamError() } resp.Body = peekedBody{Reader: io.MultiReader(&pending, br), Closer: resp.Body} @@ -297,7 +303,7 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http } if readErr != nil { stopTimer() - _ = resp.Body.Close() + closeBody() if expired.Load() { return nil, emptyUpstreamError() } diff --git a/internal/providers/qwenwork/sse.go b/internal/providers/qwenwork/sse.go index d33d515b..7466334d 100644 --- a/internal/providers/qwenwork/sse.go +++ b/internal/providers/qwenwork/sse.go @@ -11,6 +11,7 @@ import ( "sort" "strconv" "strings" + "sync" "sync/atomic" "time" @@ -613,10 +614,15 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http timeout = defaultMeaningfulUpstreamTimeout } + var closeOnce sync.Once + closeBody := func() { + closeOnce.Do(func() { _ = resp.Body.Close() }) + } + var expired atomic.Bool timer := time.AfterFunc(timeout, func() { expired.Store(true) - _ = resp.Body.Close() + closeBody() }) stopTimer := func() { _ = timer.Stop() } @@ -630,14 +636,14 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http if frame, meaningful := meaningfulDataLine(strings.TrimSpace(data)); meaningful { stopTimer() if expired.Load() { - _ = resp.Body.Close() + closeBody() return nil, emptyUpstreamError() } if frame.errorOnly() { // Nothing has reached the client yet: return the // classified error so the executor cools this account // (quota, auth, model) and fails over. - resp.Body.Close() + closeBody() return nil, frame.providerError() } resp.Body = peekedBody{Reader: io.MultiReader(&pending, br), Closer: resp.Body} @@ -647,7 +653,7 @@ func ensureMeaningfulUpstream(resp *http.Response, timeout time.Duration) (*http } if readErr != nil { stopTimer() - _ = resp.Body.Close() + closeBody() if expired.Load() { return nil, emptyUpstreamError() }