Fix tx replay attack 241 #249
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,6 +1,7 @@ | ||
| package ai | ||
|
|
||
| import ( | ||
| "bufio" | ||
|
Contributor
There was a problem hiding this comment. 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win GenerateStream duplicates Generate's request-building logic. Same DRY concern as in Also applies to: 152-188 🤖 Prompt for AI Agents |
||
| "bytes" | ||
| "context" | ||
| "encoding/json" | ||
|
|
@@ -100,3 +101,101 @@ func (p *OpenRouterProvider) Generate(ctx context.Context, text string) (string, | |
|
|
||
| return content, nil | ||
| } | ||
|
|
||
| // ErrIncompleteStream is returned by Recv when the upstream TCP connection | ||
| // closes before the [DONE] sentinel is received. This distinguishes a clean | ||
| // completion from a dropped or truncated stream. | ||
| var ErrIncompleteStream = errors.New("upstream stream ended without [DONE]") | ||
|
|
||
| type openRouterStream struct { | ||
| resp *http.Response | ||
| reader *bufio.Reader | ||
| sawDone bool | ||
| } | ||
|
|
||
| func (s *openRouterStream) Recv() (string, error) { | ||
| for { | ||
| line, err := s.reader.ReadBytes('\n') | ||
| if err != nil { | ||
| if errors.Is(err, io.EOF) { | ||
| if s.sawDone { | ||
| return "", io.EOF | ||
| } | ||
| return "", ErrIncompleteStream | ||
| } | ||
| return "", err | ||
|
Comment on lines
+118
to
+126
Contributor
There was a problem hiding this comment.
For OpenRouter, normal completion is the Useful? React with 👍 / 👎. |
||
| } | ||
| line = bytes.TrimSpace(line) | ||
| if len(line) == 0 { | ||
| continue | ||
| } | ||
| if bytes.HasPrefix(line, []byte("data: ")) { | ||
| data := bytes.TrimPrefix(line, []byte("data: ")) | ||
| if string(data) == "[DONE]" { | ||
| s.sawDone = true | ||
| return "", io.EOF | ||
| } | ||
| var chunk map[string]interface{} | ||
| if err := json.Unmarshal(data, &chunk); err != nil { | ||
| continue // skip invalid json | ||
| } | ||
| choices, ok := chunk["choices"].([]interface{}) | ||
| if !ok || len(choices) == 0 { | ||
| continue | ||
| } | ||
| choice, ok := choices[0].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| delta, ok := choice["delta"].(map[string]interface{}) | ||
| if !ok { | ||
| continue | ||
| } | ||
| if content, ok := delta["content"].(string); ok { | ||
| return content, nil | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
Comment on lines
+116
to
+159
Contributor
There was a problem hiding this comment. 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win 🧩 Analysis chain🏁 Script executed: #!/bin/bash
set -euo pipefail
# Map the relevant files first
ast-grep outline gateway/internal/ai/openrouter.go --view expanded
ast-grep outline gateway/main.go --view expanded
# Read the relevant sections with line numbers
sed -n '1,240p' gateway/internal/ai/openrouter.go | cat -n
printf '\n--- main.go ---\n'
sed -n '1,260p' gateway/main.go | cat -n
# Look for stream/error handling and transaction usage sites
rg -n "markTransactionUsed|GenerateStream|Recv\(\)|io\.EOF|openrouter" gateway -SRepository: AnkanMisra/MicroAI-Paygate Length of output: 23374 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Show the stream consumption path in main.go
sed -n '520,590p' gateway/main.go | cat -n
# Show the Ollama stream parser for comparison
sed -n '70,120p' gateway/internal/ai/ollama.go | cat -n
# Show any tests covering streaming error behavior
sed -n '1,220p' gateway/errors_test.go | cat -n
sed -n '1,180p' gateway/timeout_test.go | cat -nRepository: AnkanMisra/MicroAI-Paygate Length of output: 19102 🏁 Script executed: #!/bin/bash
set -euo pipefail
sed -n '520,590p' gateway/main.go | cat -n
printf '\n--- ollama.go ---\n'
sed -n '70,120p' gateway/internal/ai/ollama.go | cat -n
printf '\n--- errors_test.go ---\n'
sed -n '1,220p' gateway/errors_test.go | cat -n
printf '\n--- timeout_test.go ---\n'
sed -n '1,180p' gateway/timeout_test.go | cat -nRepository: AnkanMisra/MicroAI-Paygate Length of output: 19170 🏁 Script executed: #!/bin/bash
set -euo pipefail
# Show the stream consumption path in main.go
sed -n '520,590p' gateway/main.go | cat -n
# Show the Ollama stream parser for comparison
printf '\n--- ollama.go ---\n'
sed -n '70,120p' gateway/internal/ai/ollama.go | cat -n
# Show tests that may cover stream error behavior
printf '\n--- errors_test.go ---\n'
sed -n '1,220p' gateway/errors_test.go | cat -n
printf '\n--- timeout_test.go ---\n'
sed -n '1,180p' gateway/timeout_test.go | cat -nRepository: AnkanMisra/MicroAI-Paygate Length of output: 19170 🌐 Web query:
💡 Result: When a mid-stream error occurs during an OpenRouter Server-Sent Events (SSE) stream, the API communicates the error by sending a specific JSON object within the stream [1][2]. Because the HTTP status code is already set to 200 OK once streaming begins, the error cannot be signaled via standard HTTP error codes [1][3]. Instead, OpenRouter sends a final SSE chunk with a top-level error field that follows this structure [2][3]: { "id": "cmpl-abc123", "object": "chat.completion.chunk", "created": 1234567890, "model": "...", "provider": "...", "error": { "code": "server_error", "message": "Provider disconnected unexpectedly" }, "choices": [{ "index": 0, "delta": { "content": "" }, "finish_reason": "error" }] } Key behaviors include: - Termination: The stream is terminated immediately after this event is sent [1][2]. - Error Handling: Client implementations should monitor every chunk for the presence of an error field [1]. If found, the application should cease processing the stream and handle the error based on the code and message provided [1][4]. - Distinguishing Network Issues: Be aware that not all mid-stream interruptions are signaled by this JSON event [5]. If the server-side process or network connection terminates abruptly (e.g., a TLS socket closure), you may receive a low-level network error (like a "terminated" TypeError) rather than a clean JSON-formatted error chunk [5]. In such cases, there will be no response body or HTTP status code, necessitating robust retry logic for network-level failures [5]. Citations:
Surface OpenRouter SSE errors instead of treating EOF as success OpenRouter can emit mid-stream failures as an in-band chunk with a top-level 🤖 Prompt for AI Agents |
||
|
|
||
| func (s *openRouterStream) Close() error { | ||
| return s.resp.Body.Close() | ||
| } | ||
|
|
||
| func (p *OpenRouterProvider) GenerateStream(ctx context.Context, text string) (Stream, error) { | ||
| prompt := fmt.Sprintf("Summarize this text in 2 sentences: %s", text) | ||
|
|
||
| reqBody, _ := json.Marshal(map[string]interface{}{ | ||
| "model": p.model, | ||
| "messages": []map[string]string{ | ||
| {"role": "user", "content": prompt}, | ||
| }, | ||
| "stream": true, | ||
| }) | ||
|
|
||
| req, err := http.NewRequestWithContext(ctx, "POST", p.url, bytes.NewBuffer(reqBody)) | ||
| if err != nil { | ||
| return nil, fmt.Errorf("failed to create OpenRouter request: %w", err) | ||
| } | ||
| req.Header.Set("Authorization", "Bearer "+p.apiKey) | ||
| req.Header.Set("Content-Type", "application/json") | ||
|
|
||
| resp, err := http.DefaultClient.Do(req) | ||
| if err != nil { | ||
| if errors.Is(err, context.DeadlineExceeded) || ctx.Err() == context.DeadlineExceeded { | ||
| return nil, context.DeadlineExceeded | ||
| } | ||
| return nil, err | ||
| } | ||
|
|
||
| if resp.StatusCode < 200 || resp.StatusCode >= 300 { | ||
| body, _ := io.ReadAll(resp.Body) | ||
| resp.Body.Close() | ||
| return nil, fmt.Errorf("openrouter returned status %d: %s", resp.StatusCode, string(body)) | ||
| } | ||
|
|
||
| return &openRouterStream{ | ||
| resp: resp, | ||
| reader: bufio.NewReader(resp.Body), | ||
| }, nil | ||
| } | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This added SSE parser calls
strings.Split/HasPrefix/TrimSpace, butstringsis not imported in this test file; the same missing import exists ingateway/receipt_store_integration_test.gowhere the receipt stream is parsed. The gateway package will fail to compile before the replay regression can run, so addstringsto both import blocks.Useful? React with 👍 / 👎.