#415·coai

Propagate downstream stream cancellation to upstream SSE requests

Author: chenshj73Created Sep 1, 2026Updated Sep 1, 2026

Summary

Streaming chat requests are handled by a background goroutine that forwards upstream chunks through an unbuffered channel to gin.Context.Stream. If the downstream client disconnects or stops reading, the stream loop can stop while the goroutine is still trying to send the next chunk into partial.

At the same time, the upstream SSE request is created without the original Gin request context, so downstream cancellation is not propagated to the provider request.

This can leave a streaming request stuck after the client has gone away, while the upstream provider connection continues consuming resources.

Evidence

The streaming response starts a goroutine and forwards chunks into an unbuffered channel:

215 func sendStreamTranshipmentResponse(c *gin.Context, form RelayForm, messages []globals.Message, id string, created int64, user *auth.User, plan bool) {
216     partial := make(chan RelayStreamResponse)
217     db := utils.GetDBFromContext(c)
218     cache := utils.GetCacheFromContext(c)
219 
220     group := auth.GetGroup(db, user)
221     charge := channel.ChargeInstance.GetCharge(form.Model)
222 
223     go func() {
224         buffer := utils.NewBuffer(form.Model, messages, charge)
225         hit, err := channel.NewChatRequestWithCache(
226             cache, buffer, group, getChatProps(form, messages, buffer),
227             func(data *globals.Chunk) error {
228                 buffer.WriteChunk(data)
229 
230                 if !data.IsEmpty() {
231                     partial <- getStreamTranshipmentForm(id, created, form, data, buffer, false, nil)
232                 }
233                 return nil
234             },
235         )

The response loop reads from that channel and writes the SSE response:

256     c.Stream(func(w io.Writer) bool {
257         if resp, ok := <-partial; ok {
258             if resp.Error != nil {
259                 sendErrorResponse(c, resp.Error)
260                 return false
261             }
262 
263             c.Render(-1, utils.NewEvent(resp))
264             return true
265         }
266 
267         c.Render(-1, utils.NewEndEvent())
268         return false
269     })
270 }

The common SSE helper used by the OpenAI/Claude stream adapters creates the upstream request without a context:

14 type EventScannerProps struct {
15     Method   string
16     Uri      string
17     Headers  map[string]string
18     Body     interface{}
19     Callback func(string) error
20     FullSSE  bool
21 }

40 func EventScanner(props *EventScannerProps, config ...globals.ProxyConfig) *EventScannerError {
41     // panic recovery
42     defer func() {
43         if r := recover(); r != nil {
44             stack := debug.Stack()
45             globals.Warn(fmt.Sprintf("event source panic: %s (uri: %s, method: %s)\n%s", r, props.Uri, props.Method, stack))
46         }
47     }()
48 
49     if globals.DebugMode {
50         globals.Debug(fmt.Sprintf("[sse] event source: %s %s\nheaders: %v\nbody: %v", props.Method, props.Uri, Marshal(props.Headers), Marshal(props.Body)))
51     }
52 
53     client := newClient(config)
54     req, err := http.NewRequest(props.Method, props.Uri, ConvertBody(props.Body))
55     if err != nil {
56         if globals.DebugMode {
57             globals.Debug(fmt.Sprintf("[sse] failed to create request: %s", err))
58         }

OpenAI-compatible streaming calls use this helper:

127     ticks := 0
128     err := utils.EventScanner(&utils.EventScannerProps{
129         Method:  "POST",
130         Uri:     c.GetChatEndpoint(props),
131         Headers: c.GetHeader(),
132         Body:    c.GetChatBody(props, true),
133         Callback: func(data string) error {
134             ticks += 1
135 
136             partial, err := c.ProcessLine(data, isCompletionType)
137             if err != nil {
138                 return err
139             }
140             return callback(partial)
141         },
142     }, props.Proxy)

Claude streaming calls also use it:

179 // CreateStreamChatRequest is the stream request for anthropic claude
180 func (c *ChatInstance) CreateStreamChatRequest(props *adaptercommon.ChatProps, hook globals.Hook) error {
181     err := utils.EventScanner(&utils.EventScannerProps{
182         Method:  "POST",
183         Uri:     c.GetChatEndpoint(),
184         Headers: c.GetChatHeaders(),
185         Body:    c.GetChatBody(props, true),
186         Callback: func(data string) error {
187             partial, err := c.ProcessLine(data)
188             if err != nil {
189                 return err
190             }
191 
192             return hook(partial)
193         },

Why this matters

For an LLM gateway, downstream stream cancellation is common: users close browser tabs, IDE agents cancel tool calls, mobile networks drop, or clients stop once they have enough tokens. If the gateway does not propagate that cancellation upstream, the deployment can keep paying for tokens and holding provider connections after the user is gone. Under high concurrency this may also accumulate stuck goroutines around the unbuffered channel send.

Suggested fix

One possible direction:

add a Context context.Context field to EventScannerProps;
create upstream requests with http.NewRequestWithContext(ctx, ...);
pass c.Request.Context() from sendStreamTranshipmentResponse down to the adapter layer;
make sends to partial cancellation-aware, for example:

select {
case partial <- resp:
    return nil
case <-ctx.Done():
    return ctx.Err()
}

It would also be useful to add a regression test where the stream consumer disconnects before the upstream finishes, and assert that the upstream request/goroutine is cancelled promptly.

Source: coaidev/coai

View original on GitHub View discussion on GitHub