mirror of
https://github.com/Wei-Shaw/sub2api.git
synced 2026-10-15 23:22:59 +08:00
William DrakeandClaude Opus 5
b7ec3cdadf
The raw Chat Completions passthrough returned a nil error whenever the upstream
body ended, so an upstream that cut the SSE stream mid-generation surfaced as
HTTP 200 with usage 0/0: the client kept a half-finished answer, the request was
not counted as an SLA failure, and ops saw nothing. Observed against
opencode.ai/zen/go (deepseek-v4-pro): 11 of 1346 streams in 24h recorded a first
token but no final usage, alongside a concurrent `connection reset by peer`
against the upstream edge.
Track whether the upstream ever emitted a terminal signal — [DONE], a usage
chunk, or a finish_reason — and treat a stream that ends without one as an
upstream failure. Matching only [DONE] would misfire on compatible upstreams
that EOF right after their last chunk and would drop billing for tokens that
were actually delivered, so all three count.
- nothing written to the client yet: UpstreamFailoverError, retried on another
account with no client-visible half stream
- bytes already on the wire: typed openAIUpstreamStreamReadError, so the
handler appends an SSE error frame and marks the request an SLA failure
Client cancel/disconnect keeps the previous semantics — there the upstream read
error is indistinguishable from a truncation, so partial usage is still billed
rather than blamed on the account.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
87 lines
3.2 KiB
Go
87 lines
3.2 KiB
Go
package service
|
|
|
|
import (
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"strings"
|
|
)
|
|
|
|
const (
|
|
// OpenAIUpstreamHTTP2StreamErrorCode is returned to OpenAI-compatible clients
|
|
// when an upstream HTTP/2 response stream is reset after the request started.
|
|
OpenAIUpstreamHTTP2StreamErrorCode = "upstream_http2_stream_error"
|
|
OpenAIUpstreamStreamReadErrorCode = "upstream_stream_read_error"
|
|
// OpenAIUpstreamStreamTruncatedCode is returned when an upstream SSE stream
|
|
// closes *cleanly* before delivering any terminal signal. A clean EOF carries
|
|
// no transport error, so without this classification a truncated generation is
|
|
// indistinguishable from a successful one.
|
|
OpenAIUpstreamStreamTruncatedCode = "upstream_stream_truncated"
|
|
)
|
|
|
|
// ErrOpenAIUpstreamStreamTruncated marks an upstream SSE stream that ended at
|
|
// EOF — without a read error — before any terminal signal arrived.
|
|
var ErrOpenAIUpstreamStreamTruncated = errors.New("upstream stream ended before any terminal chunk")
|
|
|
|
type openAIUpstreamStreamReadError struct {
|
|
cause error
|
|
clientCode string
|
|
clientMessage string
|
|
}
|
|
|
|
func (e *openAIUpstreamStreamReadError) Error() string {
|
|
return fmt.Sprintf("stream usage incomplete: %v", e.cause)
|
|
}
|
|
|
|
func (e *openAIUpstreamStreamReadError) Unwrap() error { return e.cause }
|
|
|
|
func newOpenAIUpstreamStreamReadError(err error) error {
|
|
code, message := classifyOpenAIUpstreamStreamReadError(err)
|
|
return &openAIUpstreamStreamReadError{
|
|
cause: err,
|
|
clientCode: code,
|
|
clientMessage: message,
|
|
}
|
|
}
|
|
|
|
// shouldClassifyOpenAIUpstreamStreamReadError excludes cancellation and
|
|
// response-size enforcement from upstream retry.
|
|
func shouldClassifyOpenAIUpstreamStreamReadError(err error, contexts ...context.Context) bool {
|
|
if err == nil || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, ErrUpstreamResponseBodyTooLarge) {
|
|
return false
|
|
}
|
|
for _, ctx := range contexts {
|
|
if ctx != nil && ctx.Err() != nil {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
// OpenAIUpstreamStreamReadErrorDetails returns the stable, sanitized client
|
|
// classification attached to an upstream stream read failure.
|
|
func OpenAIUpstreamStreamReadErrorDetails(err error) (code, message string, ok bool) {
|
|
var streamErr *openAIUpstreamStreamReadError
|
|
if !errors.As(err, &streamErr) || streamErr == nil {
|
|
return "", "", false
|
|
}
|
|
return streamErr.clientCode, streamErr.clientMessage, true
|
|
}
|
|
|
|
func classifyOpenAIUpstreamStreamReadError(err error) (code, message string) {
|
|
if err != nil {
|
|
if errors.Is(err, ErrOpenAIUpstreamStreamTruncated) {
|
|
return OpenAIUpstreamStreamTruncatedCode, "Upstream response stream ended before completion"
|
|
}
|
|
lower := strings.ToLower(err.Error())
|
|
// net/http's HTTP/2 stream error is unexported. Its stable text contains
|
|
// "stream error: stream ID ..."; match only the transport signature and
|
|
// never pass the original text to the client.
|
|
if strings.Contains(lower, "stream error: stream id ") ||
|
|
(strings.Contains(lower, "http2:") && strings.Contains(lower, "stream")) {
|
|
return OpenAIUpstreamHTTP2StreamErrorCode, "Upstream HTTP/2 stream failed"
|
|
}
|
|
}
|
|
return OpenAIUpstreamStreamReadErrorCode, "Upstream response stream was interrupted"
|
|
}
|