Files
Nostalgia 893457cd50 fix(aliyundrive): limit callback concurrency (#3071)
* fix(aliyundrive): limit callback concurrency

- Share proxy callback admission by Aliyun user identity and hold permits for complete response-body lifetimes.
- Retry only verified callback-capacity rejections while preserving direct redirects and server download limiting.
- Map exhausted temporary capacity to S3 SlowDown through the merged OpenListTeam gofakes3 module.
- Cover shared limits, lifecycle release, cancellation, retry classification, and the S3 HTTP response.

Co-authored-by: Codex <267193182+codex@users.noreply.github.com>

# Conflicts:
#	go.mod
#	go.sum
#	server/s3/pager.go

* fix(op): separate redirect and proxy link cache entries

- Include redirect mode in the link cache key for all drivers.

- Cover both redirect-to-proxy and proxy-to-redirect cache reuse.

Co-authored-by: Codex <267193182+codex@users.noreply.github.com>

* fix(proxy): close range bodies before opening next

- make ServeHTTP own each range body and preserve cleanup failures
- remove the aggregate range closer and pass range readers directly
- replace the obsolete callback transport test with focused lifecycle coverage

Co-authored-by: Codex <267193182+codex@users.noreply.github.com>

---------

Co-authored-by: nostalume <nostalucent@gmail.com>
Co-authored-by: Codex <267193182+codex@users.noreply.github.com>
2026-09-24 12:01:50 +08:00

156 lines
3.8 KiB
Go

package common
import (
"context"
"fmt"
"io"
"net/http"
"strings"
"maps"
"github.com/OpenListTeam/OpenList/v4/internal/conf"
"github.com/OpenListTeam/OpenList/v4/internal/model"
"github.com/OpenListTeam/OpenList/v4/internal/net"
"github.com/OpenListTeam/OpenList/v4/internal/sign"
"github.com/OpenListTeam/OpenList/v4/internal/stream"
"github.com/OpenListTeam/OpenList/v4/pkg/utils"
)
func Proxy(w http.ResponseWriter, r *http.Request, link *model.Link, file model.Obj) error {
// if link.MFile != nil {
// attachHeader(w, file, link)
// http.ServeContent(w, r, file.GetName(), file.ModTime(), link.MFile)
// return nil
// }
if link.Concurrency > 0 || link.PartSize > 0 {
attachHeader(w, file, link)
size := link.ContentLength
if size <= 0 {
size = file.GetSize()
}
rrf, _ := stream.GetRangeReaderFromLink(size, link)
if link.RangeReader == nil {
r = r.WithContext(context.WithValue(r.Context(), conf.RequestHeaderKey, r.Header))
}
return net.ServeHTTP(w, r, file.GetName(), file.ModTime(), size, rrf)
}
if link.RangeReader != nil {
attachHeader(w, file, link)
size := link.ContentLength
if size <= 0 {
size = file.GetSize()
}
return net.ServeHTTP(w, r, file.GetName(), file.ModTime(), size, link.RangeReader)
}
//transparent proxy
header := net.ProcessHeader(r.Header, link.Header)
res, err := net.RequestHttp(r.Context(), r.Method, header, link.URL)
if err != nil {
return err
}
defer res.Body.Close()
maps.Copy(w.Header(), res.Header)
w.Header().Set("Content-Disposition", utils.GenerateContentDisposition(file.GetName()))
w.WriteHeader(res.StatusCode)
if r.Method == http.MethodHead {
return nil
}
_, err = utils.CopyWithBuffer(w, &stream.RateLimitReader{
Reader: res.Body,
Limiter: stream.ServerDownloadLimit,
Ctx: r.Context(),
})
return err
}
func attachHeader(w http.ResponseWriter, file model.Obj, link *model.Link) {
fileName := file.GetName()
w.Header().Set("Content-Disposition", utils.GenerateContentDisposition(fileName))
w.Header().Set("Content-Type", utils.GetMimeType(fileName))
size := link.ContentLength
if size <= 0 {
size = file.GetSize()
}
w.Header().Set("Etag", GetEtag(file, size))
contentType := link.Header.Get("Content-Type")
if len(contentType) > 0 {
w.Header().Set("Content-Type", contentType)
} else {
w.Header().Set("Content-Type", utils.GetMimeType(fileName))
}
}
func GetEtag(file model.Obj, size int64) string {
hash := ""
for _, v := range file.GetHash().Export() {
if v > hash {
hash = v
}
}
if len(hash) > 0 {
return fmt.Sprintf(`"%s"`, hash)
}
// 参考nginx
return fmt.Sprintf(`"%x-%x"`, file.ModTime().Unix(), size)
}
func ProxyRange(ctx context.Context, link *model.Link, size int64) *model.Link {
if link.RangeReader == nil && !strings.HasPrefix(link.URL, GetApiUrl(ctx)+"/") {
if link.ContentLength > 0 {
size = link.ContentLength
}
rrf, err := stream.GetRangeReaderFromLink(size, link)
if err == nil {
return &model.Link{
RangeReader: rrf,
ContentLength: size,
}
}
}
return link
}
type InterceptResponseWriter struct {
http.ResponseWriter
io.Writer
}
func (iw *InterceptResponseWriter) Write(p []byte) (int, error) {
return iw.Writer.Write(p)
}
type WrittenResponseWriter struct {
http.ResponseWriter
written bool
}
func (ww *WrittenResponseWriter) Write(p []byte) (int, error) {
n, err := ww.ResponseWriter.Write(p)
if !ww.written && n > 0 {
ww.written = true
}
return n, err
}
func (ww *WrittenResponseWriter) IsWritten() bool {
return ww.written
}
func GenerateDownProxyURL(storage *model.Storage, reqPath string) string {
if storage.DownProxyURL == "" {
return ""
}
query := ""
if !storage.DisableProxySign {
query = "?sign=" + sign.Sign(reqPath)
}
return fmt.Sprintf("%s%s%s",
strings.Split(storage.DownProxyURL, "\n")[0],
utils.EncodePath(reqPath, true),
query,
)
}