fix(relay): set Request.GetBody so the HTTP/2 transport can transparently retry after an upstream stream reset

common/body_storage.go

@@ -20,6 +20,13 @@ type BodyStorage interface {
 	Size() int64
 	// IsDisk 是否是磁盘存储
 	IsDisk() bool
+	// NewReader returns an independent reader positioned at the start of the
+	// stored payload. Each call returns a reader with its own cursor, so
+	// callers (e.g. http.Request.GetBody) can replay the body concurrently
+	// with, or after, other readers without sharing seek state. Closing the
+	// returned reader releases only that reader, never the storage itself;
+	// after the storage has been closed, NewReader returns ErrStorageClosed.
+	NewReader() (io.ReadCloser, error)
 }
 
 // ErrStorageClosed 存储已关闭错误
@@ -80,6 +87,18 @@ func (m *memoryStorage) Bytes() ([]byte, error) {
 	return m.data, nil
 }
 
+func (m *memoryStorage) NewReader() (io.ReadCloser, error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	if atomic.LoadInt32(&m.closed) == 1 {
+		return nil, ErrStorageClosed
+	}
+	// A fresh bytes.Reader over the shared immutable backing array: an
+	// independent cursor at zero copy cost. NopCloser keeps Close a no-op, so
+	// the storage lifecycle stays owned by whoever holds the storage itself.
+	return io.NopCloser(bytes.NewReader(m.data)), nil
+}
+
 func (m *memoryStorage) Size() int64 {
 	return m.size
 }
@@ -229,6 +248,24 @@ func (d *diskStorage) Bytes() ([]byte, error) {
 	return data, nil
 }
 
+func (d *diskStorage) NewReader() (io.ReadCloser, error) {
+	d.mu.Lock()
+	defer d.mu.Unlock()
+	if atomic.LoadInt32(&d.closed) == 1 {
+		return nil, ErrStorageClosed
+	}
+	// A separate file descriptor over the same cache file: an independent
+	// cursor at zero copy cost. Closing the returned reader closes only that
+	// descriptor; the storage keeps owning the primary descriptor and the
+	// file's lifetime. Readers opened before Close stay usable even after the
+	// file is unlinked, as the descriptor keeps the inode alive.
+	file, err := os.Open(d.filePath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to open body cache file for replay: %w", err)
+	}
+	return file, nil
+}
+
 func (d *diskStorage) Size() int64 {
 	return d.size
 }

relay/alpha_search_handler.go

@@ -62,12 +62,13 @@ func AlphaSearchHelper(c *gin.Context, info *relaycommon.RelayInfo) (newAPIError
 	}
 
 	logger.LogDebug(c, "requestBody: %s", jsonData)
-	body, size, closer, err := relaycommon.NewOutboundJSONBody(jsonData)
+	body, size, getBody, closer, err := relaycommon.NewOutboundJSONBody(jsonData)
 	if err != nil {
 		return types.NewError(err, types.ErrorCodeConvertRequestFailed, types.ErrOptionWithSkipRetry())
 	}
 	defer closer.Close()
 	info.UpstreamRequestBodySize = size
+	info.UpstreamRequestGetBody = getBody
 
 	adaptor := GetAdaptor(info.ApiType)
 	if adaptor == nil {

relay/channel/api_request.go

@@ -42,6 +42,33 @@ func applyUpstreamContentLength(req *http.Request, info *common.RelayInfo) {
 	}
 }
 
+// applyUpstreamGetBody populates req.GetBody when the upstream body is wrapped
+// in a BodyStorage (see relay/common/outbound_body.go).
+//
+// net/http.NewRequest only auto-populates GetBody for *bytes.Reader,
+// *bytes.Buffer and *strings.Reader. When the body is a type-erased io.Reader
+// (which is the case for ReaderOnly(BodyStorage)), GetBody would otherwise stay
+// nil, and the HTTP/2 transport cannot transparently retry the request once the
+// upstream resets the stream after the body was already written; the request
+// then fails with "http2: Transport: cannot retry err ... after Request.Body
+// was written; define Request.GetBody to avoid this error".
+func applyUpstreamGetBody(req *http.Request, info *common.RelayInfo) {
+	if info == nil || info.UpstreamRequestGetBody == nil {
+		return
+	}
+	if req.GetBody == nil {
+		req.GetBody = info.UpstreamRequestGetBody
+	}
+}
+
+// ApplyUpstreamBodyMetadata restores metadata that net/http cannot infer when
+// a BodyStorage is exposed through a type-erased reader. Provider adaptors
+// that construct requests directly should call this before sending them.
+func ApplyUpstreamBodyMetadata(req *http.Request, info *common.RelayInfo) {
+	applyUpstreamContentLength(req, info)
+	applyUpstreamGetBody(req, info)
+}
+
 func SetupApiRequestHeader(info *common.RelayInfo, c *gin.Context, req *http.Header) {
 	if info.RelayMode == constant.RelayModeAudioTranscription || info.RelayMode == constant.RelayModeAudioTranslation {
 		// multipart/form-data
@@ -314,7 +341,7 @@ func DoApiRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBody
 	if err != nil {
 		return nil, fmt.Errorf("new request failed: %w", err)
 	}
-	applyUpstreamContentLength(req, info)
+	ApplyUpstreamBodyMetadata(req, info)
 	headers := req.Header
 	err = a.SetupRequestHeader(c, &headers, info)
 	if err != nil {
@@ -344,7 +371,7 @@ func DoFormRequest(a Adaptor, c *gin.Context, info *common.RelayInfo, requestBod
 	if err != nil {
 		return nil, fmt.Errorf("new request failed: %w", err)
 	}
-	applyUpstreamContentLength(req, info)
+	ApplyUpstreamBodyMetadata(req, info)
 	// set form data
 	req.Header.Set("Content-Type", c.Request.Header.Get("Content-Type"))
 	headers := req.Header
@@ -474,11 +501,24 @@ func sendPingData(c *gin.Context, mutex *sync.Mutex) error {
 func DoRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http.Response, error) {
 	return doRequest(c, req, info)
 }
+
+// keepUpstreamRedirectResponse stops net/http from following redirects while
+// returning the upstream 3xx response to the relay without an extra error.
+func keepUpstreamRedirectResponse(_ *http.Request, _ []*http.Request) error {
+	return http.ErrUseLastResponse
+}
+
 func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http.Response, error) {
 	client, err := service.GetHttpClientWithProxySettings(info.ChannelSetting.Proxy, info.ChannelSetting)
 	if err != nil {
 		return nil, fmt.Errorf("new proxy http client failed: %w", err)
 	}
+	// Clients are cached and shared across channels, so override redirect
+	// behavior on a shallow copy instead of mutating the cached client. This
+	// still reuses its transport and connection pools, including HTTP/2's
+	// transparent stream retries.
+	relayClient := *client
+	relayClient.CheckRedirect = keepUpstreamRedirectResponse
 	if common2.DebugEnabled && req != nil && req.URL != nil {
 		policy := service.NormalizeHTTPTransportPolicy(info.ChannelSetting)
 		logger.LogDebug(c, fmt.Sprintf(
@@ -510,7 +550,7 @@ func doRequest(c *gin.Context, req *http.Request, info *common.RelayInfo) (*http
 		}
 	}
 
-	resp, err := client.Do(req)
+	resp, err := relayClient.Do(req)
 	if err != nil {
 		logger.LogError(c, "do request failed: "+err.Error())
 		return nil, types.NewError(err, types.ErrorCodeDoRequestFailed, types.ErrOptionWithHideErrMsg("upstream error: do request failed"))
@@ -548,10 +588,15 @@ func DoTaskApiRequest(a TaskAdaptor, c *gin.Context, info *common.RelayInfo, req
 	if err != nil {
 		return nil, fmt.Errorf("new request failed: %w", err)
 	}
-	applyUpstreamContentLength(req, info)
-	req.GetBody = func() (io.ReadCloser, error) {
-		return io.NopCloser(requestBody), nil
-	}
+	ApplyUpstreamBodyMetadata(req, info)
+	// Do NOT wrap requestBody in a GetBody closure here: returning the same
+	// (already consumed) reader would make any transport-level retry silently
+	// replay an empty body. http.NewRequest already derives a correct,
+	// snapshot-based GetBody for *bytes.Reader/Buffer/strings.Reader bodies
+	// (which most task adaptors pass in); for type-erased readers,
+	// ApplyUpstreamBodyMetadata wires a replayable body when one is available.
+	// Otherwise GetBody stays nil so the transport fails the retry instead of
+	// sending a corrupted request.
 
 	err = a.BuildRequestHeader(c, req, info)
 	if err != nil {