fix(executor): increase buffer size for stream scanners to 50MB across multiple executors

sususu committed 11 months ago

Commit 76c563d1 parent 94d61c711 files · +15 −15Browse files

Unified · Split

11 files

internal/runtime/executor/antigravity_executor.go +1 −1
internal/runtime/executor/claude_executor.go +2 −2
internal/runtime/executor/codex_executor.go +1 −1
internal/runtime/executor/gemini_cli_executor.go +1 −1
internal/runtime/executor/gemini_executor.go +1 −1
internal/runtime/executor/gemini_vertex_executor.go +2 −2
internal/runtime/executor/iflow_executor.go +1 −1
internal/runtime/executor/openai_compat_executor.go +1 −1
internal/runtime/executor/qwen_executor.go +1 −1
internal/translator/claude/gemini/claude_gemini_response.go +2 −2
internal/translator/claude/openai/responses/claude_openai-responses_response.go +2 −2

internal/runtime/executor/antigravity_executor.go+1−1

@@ -38,7 +38,7 @@ const ( · 37 unmodified lines
38	defaultAntigravityAgent = "antigravity/1.11.5 windows/amd64"	38	defaultAntigravityAgent = "antigravity/1.11.5 windows/amd64"
39	antigravityAuthType = "antigravity"	39	antigravityAuthType = "antigravity"
40	refreshSkew = 3000 * time.Second	40	refreshSkew = 3000 * time.Second
41	streamScannerBuffer int = 20_971_520	41	streamScannerBuffer int = 52_428_800 // 50MB
42	)	42	)
43		43
44	var randSource = rand.New(rand.NewSource(time.Now().UnixNano()))	44	var randSource = rand.New(rand.NewSource(time.Now().UnixNano()))

internal/runtime/executor/claude_executor.go+2−2

@@ -238,7 +238,7 @@ func (e ClaudeExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.A · 237 unmodified lines
238	// If from == to (Claude → Claude), directly forward the SSE stream without translation	238	// If from == to (Claude → Claude), directly forward the SSE stream without translation
239	if from == to {	239	if from == to {
240	scanner := bufio.NewScanner(decodedBody)	240	scanner := bufio.NewScanner(decodedBody)
241	scanner.Buffer(nil, 20_971_520)	241	scanner.Buffer(nil, 52_428_800) // 50MB
242	for scanner.Scan() {	242	for scanner.Scan() {
243	line := scanner.Bytes()	243	line := scanner.Bytes()
244	appendAPIResponseChunk(ctx, e.cfg, line)	244	appendAPIResponseChunk(ctx, e.cfg, line)
@@ -261,7 +261,7 @@ func (e ClaudeExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.A · 16 unmodified lines
261		261
262	// For other formats, use translation	262	// For other formats, use translation
263	scanner := bufio.NewScanner(decodedBody)	263	scanner := bufio.NewScanner(decodedBody)
264	scanner.Buffer(nil, 20_971_520)	264	scanner.Buffer(nil, 52_428_800) // 50MB
265	var param any	265	var param any
266	for scanner.Scan() {	266	for scanner.Scan() {
267	line := scanner.Bytes()	267	line := scanner.Bytes()

internal/runtime/executor/codex_executor.go+1−1

@@ -205,7 +205,7 @@ func (e CodexExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.Au · 204 unmodified lines
205	}	205	}
206	}()	206	}()
207	scanner := bufio.NewScanner(httpResp.Body)	207	scanner := bufio.NewScanner(httpResp.Body)
208	scanner.Buffer(nil, 20_971_520)	208	scanner.Buffer(nil, 52_428_800) // 50MB
209	var param any	209	var param any
210	for scanner.Scan() {	210	for scanner.Scan() {
211	line := scanner.Bytes()	211	line := scanner.Bytes()

internal/runtime/executor/gemini_cli_executor.go+1−1

@@ -309,7 +309,7 @@ func (e GeminiCLIExecutor) ExecuteStream(ctx context.Context, auth cliproxyaut · 308 unmodified lines
309	}()	309	}()
310	if opts.Alt == "" {	310	if opts.Alt == "" {
311	scanner := bufio.NewScanner(resp.Body)	311	scanner := bufio.NewScanner(resp.Body)
312	scanner.Buffer(nil, 20_971_520)	312	scanner.Buffer(nil, 52_428_800) // 50MB
313	var param any	313	var param any
314	for scanner.Scan() {	314	for scanner.Scan() {
315	line := scanner.Bytes()	315	line := scanner.Bytes()

internal/runtime/executor/gemini_executor.go+1−1

@@ -243,7 +243,7 @@ func (e GeminiExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.A · 242 unmodified lines
243	}	243	}
244	}()	244	}()
245	scanner := bufio.NewScanner(httpResp.Body)	245	scanner := bufio.NewScanner(httpResp.Body)
246	scanner.Buffer(nil, 20_971_520)	246	scanner.Buffer(nil, 52_428_800) // 50MB
247	var param any	247	var param any
248	for scanner.Scan() {	248	for scanner.Scan() {
249	line := scanner.Bytes()	249	line := scanner.Bytes()

internal/runtime/executor/gemini_vertex_executor.go+2−2

@@ -564,7 +564,7 @@ func (e *GeminiVertexExecutor) executeStreamWithServiceAccount(ctx context.Conte · 563 unmodified lines
564	}	564	}
565	}()	565	}()
566	scanner := bufio.NewScanner(httpResp.Body)	566	scanner := bufio.NewScanner(httpResp.Body)
567	scanner.Buffer(nil, 20_971_520)	567	scanner.Buffer(nil, 52_428_800) // 50MB
568	var param any	568	var param any
569	for scanner.Scan() {	569	for scanner.Scan() {
570	line := scanner.Bytes()	570	line := scanner.Bytes()
@@ -678,7 +678,7 @@ func (e *GeminiVertexExecutor) executeStreamWithAPIKey(ctx context.Context, auth · 107 unmodified lines
678	}	678	}
679	}()	679	}()
680	scanner := bufio.NewScanner(httpResp.Body)	680	scanner := bufio.NewScanner(httpResp.Body)
681	scanner.Buffer(nil, 20_971_520)	681	scanner.Buffer(nil, 52_428_800) // 50MB
682	var param any	682	var param any
683	for scanner.Scan() {	683	for scanner.Scan() {
684	line := scanner.Bytes()	684	line := scanner.Bytes()

internal/runtime/executor/iflow_executor.go+1−1

@@ -201,7 +201,7 @@ func (e IFlowExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.Au · 200 unmodified lines
201	}()	201	}()
202		202
203	scanner := bufio.NewScanner(httpResp.Body)	203	scanner := bufio.NewScanner(httpResp.Body)
204	scanner.Buffer(nil, 20_971_520)	204	scanner.Buffer(nil, 52_428_800) // 50MB
205	var param any	205	var param any
206	for scanner.Scan() {	206	for scanner.Scan() {
207	line := scanner.Bytes()	207	line := scanner.Bytes()

internal/runtime/executor/openai_compat_executor.go+1−1

@@ -206,7 +206,7 @@ func (e OpenAICompatExecutor) ExecuteStream(ctx context.Context, auth cliproxy · 205 unmodified lines
206	}	206	}
207	}()	207	}()
208	scanner := bufio.NewScanner(httpResp.Body)	208	scanner := bufio.NewScanner(httpResp.Body)
209	scanner.Buffer(nil, 20_971_520)	209	scanner.Buffer(nil, 52_428_800) // 50MB
210	var param any	210	var param any
211	for scanner.Scan() {	211	for scanner.Scan() {
212	line := scanner.Bytes()	212	line := scanner.Bytes()

internal/runtime/executor/qwen_executor.go+1−1

@@ -181,7 +181,7 @@ func (e QwenExecutor) ExecuteStream(ctx context.Context, auth cliproxyauth.Aut · 180 unmodified lines
181	}	181	}
182	}()	182	}()
183	scanner := bufio.NewScanner(httpResp.Body)	183	scanner := bufio.NewScanner(httpResp.Body)
184	scanner.Buffer(nil, 20_971_520)	184	scanner.Buffer(nil, 52_428_800) // 50MB
185	var param any	185	var param any
186	for scanner.Scan() {	186	for scanner.Scan() {
187	line := scanner.Bytes()	187	line := scanner.Bytes()

internal/translator/claude/gemini/claude_gemini_response.go+2−2

@@ -331,8 +331,8 @@ func ConvertClaudeResponseToGeminiNonStream(_ context.Context, modelName string, · 330 unmodified lines
331	streamingEvents := make([][]byte, 0)	331	streamingEvents := make([][]byte, 0)
332		332
333	scanner := bufio.NewScanner(bytes.NewReader(rawJSON))	333	scanner := bufio.NewScanner(bytes.NewReader(rawJSON))
334	buffer := make([]byte, 20_971_520)	334	buffer := make([]byte, 52_428_800) // 50MB
335	scanner.Buffer(buffer, 20_971_520)	335	scanner.Buffer(buffer, 52_428_800)
336	for scanner.Scan() {	336	for scanner.Scan() {
337	line := scanner.Bytes()	337	line := scanner.Bytes()
338	// log.Debug(string(line))	338	// log.Debug(string(line))

internal/translator/claude/openai/responses/claude_openai-responses_response.go+2−2

@@ -445,8 +445,8 @@ func ConvertClaudeResponseToOpenAIResponsesNonStream(_ context.Context, _ string · 444 unmodified lines
445	// Use a simple scanner to iterate through raw bytes	445	// Use a simple scanner to iterate through raw bytes
446	// Note: extremely large responses may require increasing the buffer	446	// Note: extremely large responses may require increasing the buffer
447	scanner := bufio.NewScanner(bytes.NewReader(rawJSON))	447	scanner := bufio.NewScanner(bytes.NewReader(rawJSON))
448	buf := make([]byte, 20_971_520)	448	buf := make([]byte, 52_428_800) // 50MB
449	scanner.Buffer(buf, 20_971_520)	449	scanner.Buffer(buf, 52_428_800)
450	for scanner.Scan() {	450	for scanner.Scan() {
451	line := scanner.Bytes()	451	line := scanner.Bytes()
452	if !bytes.HasPrefix(line, dataTag) {	452	if !bytes.HasPrefix(line, dataTag) {