refactor(net)!: separate range scheduling from HTTP

drivers/alias/driver.go

@@ -259,24 +259,16 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 				_ = link.Close()
 				continue
 			}
-			l := &model.Link{
-				URL:           link.URL,
-				Header:        link.Header,
-				RangeReader:   link.RangeReader,
-				Concurrency:   link.Concurrency,
-				PartSize:      link.PartSize,
-				ContentLength: link.ContentLength,
-			}
 			if d.DownloadConcurrency > 0 {
-				l.Concurrency = d.DownloadConcurrency
+				link.Concurrency = d.DownloadConcurrency
 			}
 			if d.DownloadPartSize > 0 {
-				l.PartSize = d.DownloadPartSize * utils.KB
+				link.PartSize = d.DownloadPartSize * utils.KB
 			}
-			if l.ContentLength == 0 {
-				l.ContentLength = fi.GetSize()
+			if link.ContentLength == 0 {
+				link.ContentLength = fi.GetSize()
 			}
-			rr, err := stream.GetRangeReaderFromLink(l.ContentLength, l)
+			rr, err := stream.GetRangeReaderFromLink(link.ContentLength, link)
 			if err != nil {
 				_ = link.Close()
 				continue
@@ -327,20 +319,19 @@ func (d *Alias) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 	if err != nil {
 		return nil, err
 	}
-	resultLink := link.Clone() // 复制一份，避免修改到原始link
 	if args.Redirect {
-		return resultLink, nil
+		return link, nil
 	}
 	if d.DownloadConcurrency > 0 {
-		resultLink.Concurrency = d.DownloadConcurrency
+		link.Concurrency = d.DownloadConcurrency
 	}
 	if d.DownloadPartSize > 0 {
-		resultLink.PartSize = d.DownloadPartSize * utils.KB
+		link.PartSize = d.DownloadPartSize * utils.KB
 	}
-	if resultLink.ContentLength == 0 {
-		resultLink.ContentLength = fi.GetSize()
+	if link.ContentLength == 0 {
+		link.ContentLength = fi.GetSize()
 	}
-	return resultLink, nil
+	return link, nil
 }
 
 func (d *Alias) Other(ctx context.Context, args model.OtherArgs) (interface{}, error) {
@@ -511,7 +502,7 @@ func (d *Alias) Extract(ctx context.Context, obj model.Obj, args model.ArchiveIn
 				sign.SignArchive(reqPath)),
 		}, nil
 	}
-	return link.Clone(), nil
+	return link, nil
 }
 
 func (d *Alias) ArchiveDecompress(ctx context.Context, srcObj, dstDir model.Obj, args model.ArchiveDecompressArgs) error {

drivers/chunk/driver.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	stdpath "path"
+	"slices"
 	"strconv"
 	"strings"
 
@@ -314,7 +315,7 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 		if err != nil {
 			return nil, err
 		}
-		return l.Clone(), nil
+		return l, nil
 	}
 	// 检查0号块不等于-1 以支持空文件
 	// 如果块数量大于1 最后一块不可能为0
@@ -328,7 +329,7 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 		}
 	}
 	fileSize := chunkFile.GetSize()
-	mergedRrf := func(ctx context.Context, httpRange http_range.Range) (io.ReadCloser, error) {
+	mergedRrf := func(ctx context.Context, httpRange http_range.Range) (_ io.ReadCloser, err error) {
 		start := httpRange.Start
 		length := httpRange.Length
 		if length < 0 || start+length > fileSize {
@@ -339,6 +340,12 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 		}
 		rs := make([]io.Reader, 0)
 		cs := make(utils.Closers, 0)
+		defer func() {
+			slices.Reverse(cs)
+			if err != nil {
+				err = errors.Join(err, cs.Close())
+			}
+		}()
 		var (
 			rc       io.ReadCloser
 			readFrom bool
@@ -347,7 +354,6 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 			if readFrom {
 				l, o, err := op.Link(ctx, remoteStorage, stdpath.Join(remoteActualPath, d.getPartName(idx)), args)
 				if err != nil {
-					_ = cs.Close()
 					return nil, err
 				}
 				cs = append(cs, l)
@@ -356,12 +362,10 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 					chunkSize2 = o.GetSize()
 				}
 				if chunkSize2 != chunkSize {
-					_ = cs.Close()
 					return nil, fmt.Errorf("chunk part[%d] size not match", idx)
 				}
 				rrf, err := stream.GetRangeReaderFromLink(chunkSize2, l)
 				if err != nil {
-					_ = cs.Close()
 					return nil, err
 				}
 				newLength := length - chunkSize2
@@ -372,7 +376,9 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 					rc, err = rrf.RangeRead(ctx, http_range.Range{Length: length})
 				}
 				if err != nil {
-					_ = cs.Close()
+					if rc != nil {
+						err = errors.Join(err, rc.Close())
+					}
 					return nil, err
 				}
 				rs = append(rs, rc)
@@ -388,7 +394,6 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 			} else {
 				l, o, err := op.Link(ctx, remoteStorage, stdpath.Join(remoteActualPath, d.getPartName(idx)), args)
 				if err != nil {
-					_ = cs.Close()
 					return nil, err
 				}
 				cs = append(cs, l)
@@ -397,17 +402,17 @@ func (d *Chunk) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (
 					chunkSize2 = o.GetSize()
 				}
 				if chunkSize2 != chunkSize {
-					_ = cs.Close()
 					return nil, fmt.Errorf("chunk part[%d] size not match", idx)
 				}
 				rrf, err := stream.GetRangeReaderFromLink(chunkSize2, l)
 				if err != nil {
-					_ = cs.Close()
 					return nil, err
 				}
 				rc, err = rrf.RangeRead(ctx, http_range.Range{Start: start, Length: -1})
 				if err != nil {
-					_ = cs.Close()
+					if rc != nil {
+						err = errors.Join(err, rc.Close())
+					}
 					return nil, err
 				}
 				length -= chunkSize2 - start

drivers/crypt/driver.go

@@ -283,10 +283,11 @@ func (d *Crypt) Link(ctx context.Context, file model.Obj, _ model.LinkArgs) (*mo
 			n, err := io.ReadFull(remoteReader, fileHeader)
 			if n != fileHeaderSize {
 				fileHeader = nil
+				_ = remoteReader.Close()
 				return nil, fmt.Errorf("failed to read all data: (expect =%d, actual =%d) %w", fileHeaderSize, n, err)
 			}
 			if limit <= fileHeaderSize {
-				remoteReader.Close()
+				_ = remoteReader.Close()
 				return io.NopCloser(bytes.NewReader(fileHeader[:limit])), nil
 			} else {
 				remoteReader = utils.ReadCloser{

drivers/strm/driver.go

@@ -221,7 +221,7 @@ func (d *Strm) Link(ctx context.Context, file model.Obj, args model.LinkArgs) (*
 		}, nil
 	}
 
-	return link.Clone(), nil
+	return link, nil
 }
 
 var _ driver.Driver = (*Strm)(nil)

drivers/strm/hook.go

@@ -124,11 +124,14 @@ func generateStrm(ctx context.Context, driver *Strm, obj model.Obj, localPath st
 	}
 	rc, err := rrf.RangeRead(ctx, http_range.Range{Length: -1})
 	if err != nil {
+		if rc != nil {
+			_ = rc.Close()
+		}
 		log.Warnf("failed to generate strm of obj %s: failed to read range: %v", localPath, err)
 		return
 	}
-	defer rc.Close()
 	same, err := isSameContent(localPath, size, rc)
+	_ = rc.Close()
 	if err != nil {
 		log.Warnf("failed to compare content of obj %s: %v", localPath, err)
 		return
@@ -138,6 +141,9 @@ func generateStrm(ctx context.Context, driver *Strm, obj model.Obj, localPath st
 	}
 	rc, err = rrf.RangeRead(ctx, http_range.Range{Length: -1})
 	if err != nil {
+		if rc != nil {
+			_ = rc.Close()
+		}
 		log.Warnf("failed to generate strm of obj %s: failed to reread range: %v", localPath, err)
 		return
 	}

internal/fs/copy_move.go

@@ -254,7 +254,6 @@ func (t *FileTransferTask) RunWithNextTaskCallback(f func(nextTask *FileTransfer
 		Ctx: t.Ctx(),
 	}, link)
 	if err != nil {
-		_ = link.Close()
 		return errors.WithMessagef(err, "failed get [%s] stream", t.SrcActualPath)
 	}
 	t.SetTotalBytes(ss.GetSize())

internal/hybrid_cache/buffer.go

@@ -3,33 +3,33 @@ package hybrid_cache
 import (
 	"fmt"
 	"io"
+	"sync"
 )
 
 type BufferStore struct {
+	mu     sync.RWMutex
 	blocks [][]byte
 	size   int64
 }
 
 func (m *BufferStore) Size() int64 {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
 	return m.size
 }
 
-// 用于存储不复用的[]byte
-func (m *BufferStore) Append(buf []byte) {
-	m.size += int64(len(buf))
-	m.blocks = append(m.blocks, buf)
-}
-
 func (m *BufferStore) Close() error {
-	if len(m.blocks) > 0 {
-		clear(m.blocks)
-		m.blocks = m.blocks[:0]
-		m.size = 0
-	}
+	m.mu.Lock()
+	defer m.mu.Unlock()
+	clear(m.blocks)
+	m.blocks = nil
+	m.size = 0
 	return nil
 }
 
 func (m *BufferStore) ReadAt(p []byte, off int64) (int, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
 	if len(p) == 0 {
 		return 0, nil
 	}
@@ -55,6 +55,8 @@ func (m *BufferStore) ReadAt(p []byte, off int64) (int, error) {
 }
 
 func (m *BufferStore) WriteAt(p []byte, off int64) (int, error) {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
 	if len(p) == 0 {
 		return 0, nil
 	}
@@ -80,6 +82,8 @@ func (m *BufferStore) WriteAt(p []byte, off int64) (int, error) {
 }
 
 func (m *BufferStore) GrowTo(size int64) (err error) {
+	m.mu.Lock()
+	defer m.mu.Unlock()
 	if size <= m.size {
 		return nil
 	}

internal/hybrid_cache/buffer_test.go

@@ -14,9 +14,9 @@ func TestBufferStore(t *testing.T) {
 		off int64
 	}
 	bs := &hybrid_cache.BufferStore{}
-	bs.Append([]byte("github.com"))
-	bs.Append([]byte("/OpenList"))
-	bs.Append([]byte("Team/?"))
+	initial := []byte("github.com/OpenListTeam/?")
+	_ = bs.GrowTo(int64(len(initial)))
+	_, _ = bs.WriteAt(initial, 0)
 	b := []byte("OpenList")
 	off := bs.Size() - 1
 	_ = bs.GrowTo(off + int64(len(b)))

internal/model/args.go

@@ -30,7 +30,7 @@ type Link struct {
 	Header      http.Header   `json:"header"` // needed header (for url)
 	RangeReader RangeReaderIF `json:"-"`      // recommended way if can't use URL
 
-	Expiration *time.Duration // local cache expiration; not transferred by Clone
+	Expiration *time.Duration // local cache expiration
 
 	//for accelerating request, use multi-thread downloading
 	Concurrency   int   `json:"concurrency"`
@@ -42,12 +42,18 @@ type Link struct {
 	RequireReference bool `json:"-"`
 }
 
-// Clone transfers ownership of l without inheriting its cache expiration.
+// Clone transfers ownership of l while isolating its mutable transport and cache metadata.
 func (l *Link) Clone() *Link {
+	var expiration *time.Duration
+	if l.Expiration != nil {
+		value := *l.Expiration
+		expiration = &value
+	}
 	return &Link{
 		URL:              l.URL,
-		Header:           l.Header,
+		Header:           l.Header.Clone(),
 		RangeReader:      l.RangeReader,
+		Expiration:       expiration,
 		Concurrency:      l.Concurrency,
 		PartSize:         l.PartSize,
 		ContentLength:    l.ContentLength,