From 0310c63c795faf7c63d260e537014b4ddd53fd20 Mon Sep 17 00:00:00 2001 From: Clansty Date: Sat, 28 Mar 2026 02:55:46 +0800 Subject: [PATCH] fix: propagate client request context to upstream HTTP requests When a client disconnects mid-request, the upstream HTTP request was not being cancelled because http.NewRequest() was used instead of http.NewRequestWithContext(). This caused upstream requests to continue consuming resources until they completed or timed out. Changes: - Use http.NewRequestWithContext(c.Request.Context(), ...) in core relay functions (DoApiRequest, DoFormRequest, DoTaskApiRequest) and channel adaptors (jimeng, coze, dify, replicate) so upstream requests are automatically cancelled when the client disconnects. - Derive StreamScannerHandler's internal context from c.Request.Context() instead of context.Background(), removing redundant c.Request.Context() Done() checks that are now covered by the parent context chain. - Derive startPingKeepAlive's context from c.Request.Context() instead of context.Background() for the same reason. --- relay/channel/api_request.go | 19 ++++++++++++------- relay/channel/coze/relay-coze.go | 4 ++-- relay/channel/dify/relay-dify.go | 2 +- relay/channel/jimeng/adaptor.go | 2 +- relay/channel/replicate/adaptor.go | 2 +- relay/helper/stream_scanner.go | 11 +++-------- 6 files changed, 20 insertions(+), 20 deletions(-) diff --git a/relay/channel/api_request.go b/relay/channel/api_request.go index 8dfb61d40093..22c8579bde5f 100644 --- a/relay/channel/api_request.go +++ b/relay/channel/api_request.go @@ -295,7 +295,7 @@ func DoApiRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBody if common2.DebugEnabled { println("fullRequestURL:", fullRequestURL) } - req, err := http.NewRequest(c.Request.Method, fullRequestURL, requestBody) + req, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, fullRequestURL, requestBody) if err != nil { return nil, fmt.Errorf("new request failed: %w", err) } @@ -326,7 +326,7 @@ func DoFormRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBod if common2.DebugEnabled { println("fullRequestURL:", fullRequestURL) } - req, err := http.NewRequest(c.Request.Method, fullRequestURL, requestBody) + req, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, fullRequestURL, requestBody) if err != nil { return nil, fmt.Errorf("new request failed: %w", err) } @@ -382,7 +382,7 @@ func DoWssRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBody } func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.CancelFunc { - pingerCtx, stopPinger := context.WithCancel(context.Background()) + pingerCtx, stopPinger := context.WithCancel(c.Request.Context()) gopool.Go(func() { defer func() { @@ -433,9 +433,6 @@ func startPingKeepAlive(c *gin.Context, pingInterval time.Duration) context.Canc // 收到退出信号 case <-pingerCtx.Done(): return - // request 结束 - case <-c.Request.Context().Done(): - return // 超时保护，防止goroutine无限运行 case <-pingTimeout.C: if common2.DebugEnabled { @@ -517,6 +514,14 @@ func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http resp, err := client.Do(req) if err != nil { + // 区分客户端主动断开和真正的上游错误 + // 客户端断开时不记录错误日志、不重试、不影响渠道状态 + if c.Request.Context().Err() != nil { + logger.LogInfo(c, "request cancelled by client: "+err.Error()) + return nil, types.NewError(err, types.ErrorCodeDoRequestFailed, + types.ErrOptionWithSkipRetry(), + types.ErrOptionWithHideErrMsg("client disconnected")) + } logger.LogError(c, "do request failed: "+err.Error()) return nil, types.NewError(err, types.ErrorCodeDoRequestFailed, types.ErrOptionWithHideErrMsg("upstream error: do request failed")) } @@ -534,7 +539,7 @@ func DoTaskApiRequest(a TaskAdaptor, c *gin.Context, info *common.RelayInfo, req if err != nil { return nil, err } - req, err := http.NewRequest(c.Request.Method, fullRequestURL, requestBody) + req, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, fullRequestURL, requestBody) if err != nil { return nil, fmt.Errorf("new request failed: %w", err) } diff --git a/relay/channel/coze/relay-coze.go b/relay/channel/coze/relay-coze.go index 69ebd8a684c2..4ad773df65bb 100644 --- a/relay/channel/coze/relay-coze.go +++ b/relay/channel/coze/relay-coze.go @@ -218,7 +218,7 @@ func checkIfChatComplete(a *Adaptor, c *gin.Context, info *relaycommon.RelayInfo requestURL = requestURL + "?conversation_id=" + c.GetString("coze_conversation_id") + "&chat_id=" + c.GetString("coze_chat_id") // 将 conversationId和chatId作为参数发送get请求 - req, err := http.NewRequest("GET", requestURL, nil) + req, err := http.NewRequestWithContext(c.Request.Context(), "GET", requestURL, nil) if err != nil { return err, false } @@ -263,7 +263,7 @@ func getChatDetail(a *Adaptor, c *gin.Context, info *relaycommon.RelayInfo) (*ht requestURL := fmt.Sprintf("%s/v3/chat/message/list", info.ChannelBaseUrl) requestURL = requestURL + "?conversation_id=" + c.GetString("coze_conversation_id") + "&chat_id=" + c.GetString("coze_chat_id") - req, err := http.NewRequest("GET", requestURL, nil) + req, err := http.NewRequestWithContext(c.Request.Context(), "GET", requestURL, nil) if err != nil { return nil, fmt.Errorf("new request failed: %w", err) } diff --git a/relay/channel/dify/relay-dify.go b/relay/channel/dify/relay-dify.go index bec135b8765e..c5b86c588bf1 100644 --- a/relay/channel/dify/relay-dify.go +++ b/relay/channel/dify/relay-dify.go @@ -88,7 +88,7 @@ func uploadDifyFile(c *gin.Context, info *relaycommon.RelayInfo, user string, me writer.Close() // Create HTTP request - req, err := http.NewRequest("POST", uploadUrl, body) + req, err := http.NewRequestWithContext(c.Request.Context(), "POST", uploadUrl, body) if err != nil { common.SysLog("failed to create request: " + err.Error()) return nil diff --git a/relay/channel/jimeng/adaptor.go b/relay/channel/jimeng/adaptor.go index 1938ac1bec18..586e7413fb94 100644 --- a/relay/channel/jimeng/adaptor.go +++ b/relay/channel/jimeng/adaptor.go @@ -108,7 +108,7 @@ func (a *Adaptor) DoRequest(c *gin.Context, info *relaycommon.RelayInfo, request if err != nil { return nil, fmt.Errorf("get request url failed: %w", err) } - req, err := http.NewRequest(c.Request.Method, fullRequestURL, requestBody) + req, err := http.NewRequestWithContext(c.Request.Context(), c.Request.Method, fullRequestURL, requestBody) if err != nil { return nil, fmt.Errorf("new request failed: %w", err) } diff --git a/relay/channel/replicate/adaptor.go b/relay/channel/replicate/adaptor.go index 673502054b45..64a7f2bba97d 100644 --- a/relay/channel/replicate/adaptor.go +++ b/relay/channel/replicate/adaptor.go @@ -471,7 +471,7 @@ func uploadFileFromForm(c *gin.Context, info *relaycommon.RelayInfo, fieldCandid } uploadURL := relaycommon.GetFullRequestURL(baseURL, "/v1/files", info.ChannelType) - req, err := http.NewRequest(http.MethodPost, uploadURL, &body) + req, err := http.NewRequestWithContext(c.Request.Context(), http.MethodPost, uploadURL, &body) if err != nil { return "", fmt.Errorf("replicate adaptor: create upload request failed: %w", err) } diff --git a/relay/helper/stream_scanner.go b/relay/helper/stream_scanner.go index ae70f53c03b1..9fd81162ec38 100644 --- a/relay/helper/stream_scanner.go +++ b/relay/helper/stream_scanner.go @@ -108,7 +108,7 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon scanner.Split(bufio.ScanLines) SetEventStreamHeaders(c) - ctx, cancel := context.WithCancel(context.Background()) + ctx, cancel := context.WithCancel(c.Request.Context()) defer cancel() ctx = context.WithValue(ctx, "stop_chan", stopChan) @@ -165,9 +165,6 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon return case <-stopChan: return - case <-c.Request.Context().Done(): - // 监听客户端断开连接 - return case <-pingTimeout.C: logger.LogError(c, "ping goroutine max duration reached") return @@ -219,8 +216,6 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon return case <-ctx.Done(): return - case <-c.Request.Context().Done(): - return default: } @@ -276,8 +271,8 @@ func StreamScannerHandler(c *gin.Context, resp *http.Response, info *relaycommon case <-stopChan: // 正常结束 logger.LogInfo(c, "streaming finished") - case <-c.Request.Context().Done(): - // 客户端断开连接 + case <-ctx.Done(): + // 客户端断开连接或 context 被取消 logger.LogInfo(c, "client disconnected") } }