Skip to content

Commit f0adfac

Browse files
authored
fix(models/api): cap non-streaming provider response bodies (#4397)
Endpoint.Do bounds the error body at 1 MiB. Thirty lines below it, roundTrip read the success body whole, and the four chat protocols each did the same in their own read: a reply was buffered without limit before anything inspected it. The older models/chat layer already caps the same replies at 20 MiB. Cap them through one helper, and return an over-cap body without the TransportError wrapper. withRetry retries only TransportError, and repeating an oversized reply would bill the request again for the same bytes - the same reason its own doc gives for not retrying a reply that arrives whole but does not decode. 64 MiB clears every payload the catalog can legitimately return: an embeddings batch is bounded by BATCH_EMBED_SIZE (default 5) times the model dimensions, and a chat reply by the model's output-token ceiling. Streaming replies are assembled incrementally and are unaffected.
1 parent 7705f77 commit f0adfac

7 files changed

Lines changed: 204 additions & 5 deletions

File tree

‎internal/models/api/anthropicmessages/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -118,7 +118,7 @@ func (c *Client) Chat(ctx context.Context, messages []api.Message, opts *api.Opt
118118
}
119119
}
120120
defer func() { _ = resp.body.Close() }()
121-
raw, err := io.ReadAll(resp.body)
121+
raw, err := api.ReadResponseCapped(resp.body, 0)
122122
if err != nil {
123123
return nil, fmt.Errorf("read response: %w", err)
124124
}

‎internal/models/api/endpoint.go‎

Lines changed: 9 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import (
44
"bytes"
55
"context"
66
"encoding/json"
7+
"errors"
78
"fmt"
89
"io"
910
"net/http"
@@ -172,8 +173,15 @@ func (e Endpoint) roundTrip(req *http.Request, out any) error {
172173
return err
173174
}
174175
defer func() { _ = resp.Body.Close() }()
175-
raw, err := io.ReadAll(resp.Body)
176+
raw, err := ReadResponseCapped(resp.Body, 0)
176177
if err != nil {
178+
// A body over the cap is the vendor answering with something we cannot
179+
// use, not the network failing. withRetry retries only TransportError,
180+
// and repeating an oversized reply would bill the request again for the
181+
// same bytes, so it is returned as-is.
182+
if errors.Is(err, ErrResponseTooLarge) {
183+
return err
184+
}
177185
// The connection broke before the reply arrived whole: nothing was
178186
// received, so this is the network failing, not the vendor answering.
179187
return &TransportError{Op: "read response", Err: err}

‎internal/models/api/googlegenai/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -117,7 +117,7 @@ func (c *Client) Chat(ctx context.Context, messages []api.Message, opts *api.Opt
117117
}
118118
}
119119
defer func() { _ = resp.Body.Close() }()
120-
raw, err := io.ReadAll(resp.Body)
120+
raw, err := api.ReadResponseCapped(resp.Body, 0)
121121
if err != nil {
122122
return nil, fmt.Errorf("read response: %w", err)
123123
}

‎internal/models/api/openaicompletions/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ func (c *Client) Chat(ctx context.Context, messages []api.Message, opts *api.Opt
8484
}
8585
}
8686
defer func() { _ = resp.Body.Close() }()
87-
raw, err := io.ReadAll(resp.Body)
87+
raw, err := api.ReadResponseCapped(resp.Body, 0)
8888
if err != nil {
8989
return nil, fmt.Errorf("read response: %w", err)
9090
}

‎internal/models/api/openairesponses/client.go‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -84,7 +84,7 @@ func (c *Client) Chat(ctx context.Context, messages []api.Message, opts *api.Opt
8484
}
8585
}
8686
defer func() { _ = resp.Body.Close() }()
87-
raw, err := io.ReadAll(resp.Body)
87+
raw, err := api.ReadResponseCapped(resp.Body, 0)
8888
if err != nil {
8989
return nil, fmt.Errorf("read response: %w", err)
9090
}
Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
package api
2+
3+
import (
4+
"errors"
5+
"fmt"
6+
"io"
7+
)
8+
9+
// MaxResponseBytes bounds a single non-streaming provider response body.
10+
//
11+
// Streaming replies are assembled incrementally and are not covered by this. It
12+
// has to clear every non-streaming payload the catalog can legitimately return:
13+
// an embeddings batch is bounded by BATCH_EMBED_SIZE (default 5) times the
14+
// model's dimensions, and a chat reply by the model's output-token ceiling, so
15+
// both sit far below this even at their largest. It is also well above the 20 MiB
16+
// the older models/chat layer applies to the same replies, so nothing that
17+
// worked before can be refused by it.
18+
const MaxResponseBytes = 64 << 20
19+
20+
// ErrResponseTooLarge marks a body over the cap. It is deterministic: the same
21+
// request returns the same oversized body, so a retry cannot help.
22+
var ErrResponseTooLarge = errors.New("response exceeds maximum size")
23+
24+
// ReadResponseCapped reads a non-streaming response body, refusing anything
25+
// larger than limit instead of buffering it. Oversized payloads are reported as
26+
// an error rather than truncated, because a truncated body would otherwise be
27+
// decoded as if it were the whole reply - producing a short answer or a
28+
// plausible-looking partial embedding instead of an error. A non-positive limit
29+
// falls back to MaxResponseBytes.
30+
//
31+
// Endpoint.Do already bounds the error body; this is the same guard for the
32+
// success body, which until now was read whole.
33+
func ReadResponseCapped(body io.Reader, limit int64) ([]byte, error) {
34+
if limit <= 0 {
35+
limit = MaxResponseBytes
36+
}
37+
data, err := io.ReadAll(io.LimitReader(body, limit+1))
38+
if err != nil {
39+
return nil, err
40+
}
41+
if int64(len(data)) > limit {
42+
return nil, fmt.Errorf("%w (%d bytes)", ErrResponseTooLarge, limit)
43+
}
44+
return data, nil
45+
}
Lines changed: 146 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,146 @@
1+
package api
2+
3+
import (
4+
"context"
5+
"errors"
6+
"io"
7+
"net/http"
8+
"net/http/httptest"
9+
"strings"
10+
"testing"
11+
"time"
12+
13+
secutils "github.com/Tencent/WeKnora/internal/utils"
14+
)
15+
16+
// fillerReader streams size bytes without allocating them, so a test can offer
17+
// a body far larger than the cap without spending the memory.
18+
type fillerReader struct {
19+
remaining int64
20+
}
21+
22+
func (f *fillerReader) Read(p []byte) (int, error) {
23+
if f.remaining <= 0 {
24+
return 0, io.EOF
25+
}
26+
if int64(len(p)) > f.remaining {
27+
p = p[:f.remaining]
28+
}
29+
for i := range p {
30+
p[i] = 'a'
31+
}
32+
f.remaining -= int64(len(p))
33+
return len(p), nil
34+
}
35+
36+
// allowLoopback whitelists the loopback host Endpoint.newPost's SSRF check
37+
// rejects, so a test can point an Endpoint at its own httptest server.
38+
func allowLoopback(t *testing.T) {
39+
t.Helper()
40+
t.Setenv("SSRF_WHITELIST", "127.0.0.1,localhost")
41+
t.Setenv("SSRF_WHITELIST_EXTRA", "")
42+
secutils.ResetSSRFWhitelistForTest()
43+
t.Cleanup(secutils.ResetSSRFWhitelistForTest)
44+
}
45+
46+
func TestReadResponseCapped(t *testing.T) {
47+
t.Run("rejects over the cap", func(t *testing.T) {
48+
const limit = 1 << 10
49+
_, err := ReadResponseCapped(&fillerReader{remaining: limit + 1}, limit)
50+
if !errors.Is(err, ErrResponseTooLarge) {
51+
t.Fatalf("err = %v, want ErrResponseTooLarge", err)
52+
}
53+
if !strings.Contains(err.Error(), "exceeds maximum size") {
54+
t.Errorf("error text = %q, want the canonical over-size message", err)
55+
}
56+
})
57+
58+
t.Run("accepts exactly the cap", func(t *testing.T) {
59+
const limit = 1 << 10
60+
got, err := ReadResponseCapped(strings.NewReader(strings.Repeat("a", limit)), limit)
61+
if err != nil {
62+
t.Fatalf("ReadResponseCapped() = %v, want nil at exactly the cap", err)
63+
}
64+
if len(got) != limit {
65+
t.Errorf("len = %d, want %d", len(got), limit)
66+
}
67+
})
68+
69+
t.Run("non-positive limit falls back to the production cap", func(t *testing.T) {
70+
if _, err := ReadResponseCapped(strings.NewReader(`{"ok":true}`), 0); err != nil {
71+
t.Fatalf("ReadResponseCapped(_, 0) = %v, want nil", err)
72+
}
73+
})
74+
}
75+
76+
// The cap must not silently truncate: a short read that still parses would look
77+
// like a successful reply with fewer fields than the vendor sent.
78+
func TestReadResponseCappedNeverTruncates(t *testing.T) {
79+
const limit = 1 << 10
80+
got, err := ReadResponseCapped(strings.NewReader(strings.Repeat("a", limit*4)), limit)
81+
if err == nil {
82+
t.Fatalf("expected refusal, got %d bytes", len(got))
83+
}
84+
if got != nil {
85+
t.Errorf("returned %d bytes alongside the error; a caller could decode them", len(got))
86+
}
87+
}
88+
89+
// PostJSON is the whole round-trip for rerank, embeddings and transcription. An
90+
// oversized success body has to surface as an error rather than a decode
91+
// failure on a truncated payload.
92+
//
93+
// Endpoint.newPost validates the URL, so the test server's loopback address has
94+
// to be whitelisted the same way other tests in this repo reach an httptest
95+
// server.
96+
func TestPostJSONRejectsOversizedBody(t *testing.T) {
97+
allowLoopback(t)
98+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
99+
w.Header().Set("Content-Type", "application/json")
100+
w.WriteHeader(http.StatusOK)
101+
_, _ = w.Write([]byte(`{"data":[`))
102+
_, _ = io.Copy(w, &fillerReader{remaining: MaxResponseBytes + (1 << 20)})
103+
}))
104+
defer srv.Close()
105+
106+
var out map[string]any
107+
err := Endpoint{Client: srv.Client()}.PostJSON(
108+
context.Background(), srv.URL, []byte(`{}`), &out)
109+
if err == nil {
110+
t.Fatal("expected an oversized provider body to be refused")
111+
}
112+
if !errors.Is(err, ErrResponseTooLarge) {
113+
t.Errorf("err = %v, want ErrResponseTooLarge", err)
114+
}
115+
if out != nil {
116+
t.Errorf("out = %v, want nothing decoded from a refused body", out)
117+
}
118+
}
119+
120+
// withRetry retries only TransportError. An oversized reply is the same answer
121+
// every time, and retrying it bills the request again for the same bytes, so it
122+
// must not be classified as a transport failure.
123+
func TestOversizedBodyIsNotRetried(t *testing.T) {
124+
allowLoopback(t)
125+
126+
var attempts int
127+
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
128+
attempts++
129+
w.WriteHeader(http.StatusOK)
130+
_, _ = w.Write([]byte(`{"data":[`))
131+
_, _ = io.Copy(w, &fillerReader{remaining: MaxResponseBytes + (1 << 20)})
132+
}))
133+
defer srv.Close()
134+
135+
policy := RetryPolicy{MaxRetries: 3, BaseDelay: time.Millisecond, MaxDelay: time.Millisecond}
136+
var out map[string]any
137+
err := Endpoint{Client: srv.Client()}.PostJSONWithRetry(
138+
context.Background(), srv.URL, []byte(`{}`), &out, policy, "embedding")
139+
if err == nil {
140+
t.Fatal("expected an oversized provider body to be refused")
141+
}
142+
if attempts != 1 {
143+
t.Errorf("made %d attempts, want 1: an oversized reply is deterministic "+
144+
"and retrying it bills the request again", attempts)
145+
}
146+
}

0 commit comments

Comments
 (0)