refactor(op): isolate cached link borrowers

drivers/alias/driver.go

@@ -327,20 +327,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 +510,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

@@ -314,7 +314,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

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)

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,

internal/op/archive.go

@@ -390,9 +390,24 @@ func ArchiveGet(ctx context.Context, storage driver.Driver, path string, args mo
 }
 
 type objWithLink struct {
-	link   *model.Link
-	obj    model.Obj
-	policy linkCachePolicy
+	link *model.Link
+	obj  model.Obj
+}
+
+var errConflictingLinkLifecycle = stderrors.New("invalid link lifecycle: expiration cannot be combined with owned closers or RequireReference")
+
+func admitLink(link *model.Link, obj model.Obj) (*objWithLink, error) {
+	if link.Expiration != nil && (link.RequireReference || link.SyncClosers.Length() > 0) {
+		return nil, stderrors.Join(errConflictingLinkLifecycle, link.Close())
+	}
+	return &objWithLink{link: link, obj: obj}, nil
+}
+
+func (ol *objWithLink) acquire() *model.Link {
+	if ol.link.Expiration != nil || ol.link.SyncClosers.AcquireReference() || !ol.link.RequireReference {
+		return ol.link.Clone()
+	}
+	return nil
 }
 
 var (
@@ -406,8 +421,8 @@ func DriverExtract(ctx context.Context, storage driver.Driver, path string, args
 	}
 	key := stdpath.Join(Key(storage, path), args.InnerPath)
 	if ol, ok := extractCache.Get(key); ok {
-		if ol.acquire() {
-			return ol.link, ol.obj, nil
+		if link := ol.acquire(); link != nil {
+			return link, ol.obj, nil
 		}
 	}
 
@@ -416,8 +431,8 @@ func DriverExtract(ctx context.Context, storage driver.Driver, path string, args
 		if err != nil {
 			return nil, errors.Wrapf(err, "failed extract archive")
 		}
-		if ol.policy.expiration != nil {
-			extractCache.SetWithTTL(key, ol, *ol.policy.expiration)
+		if ol.link.Expiration != nil {
+			extractCache.SetWithTTL(key, ol, *ol.link.Expiration)
 		} else {
 			extractCache.SetWithExpirable(key, ol, &ol.link.SyncClosers)
 		}
@@ -429,8 +444,8 @@ func DriverExtract(ctx context.Context, storage driver.Driver, path string, args
 		if err != nil {
 			return nil, nil, err
 		}
-		if ol.acquire() {
-			return ol.link, ol.obj, nil
+		if link := ol.acquire(); link != nil {
+			return link, ol.obj, nil
 		}
 	}
 }

internal/op/fs.go

@@ -245,8 +245,8 @@ func Link(ctx context.Context, storage driver.Driver, path string, args model.Li
 	}
 	key := Key(storage, path)
 	if ol, exists := Cache.linkCache.GetType(key, typeKey); exists {
-		if ol.acquire() {
-			return ol.link, ol.obj, nil
+		if link := ol.acquire(); link != nil {
+			return link, ol.obj, nil
 		}
 	}
 
@@ -267,8 +267,8 @@ func Link(ctx context.Context, storage driver.Driver, path string, args model.Li
 		if err != nil {
 			return nil, err
 		}
-		if ol.policy.expiration != nil {
-			Cache.linkCache.SetTypeWithTTL(key, typeKey, ol, *ol.policy.expiration)
+		if ol.link.Expiration != nil {
+			Cache.linkCache.SetTypeWithTTL(key, typeKey, ol, *ol.link.Expiration)
 		} else {
 			Cache.linkCache.SetTypeWithExpirable(key, typeKey, ol, &link.SyncClosers)
 		}
@@ -279,8 +279,8 @@ func Link(ctx context.Context, storage driver.Driver, path string, args model.Li
 		if err != nil {
 			return nil, nil, err
 		}
-		if ol.acquire() {
-			return ol.link, ol.obj, nil
+		if link := ol.acquire(); link != nil {
+			return link, ol.obj, nil
 		}
 	}
 }

internal/op/link_lifecycle_test.go

@@ -2,6 +2,7 @@ package op
 
 import (
 	"context"
+	"net/http"
 	"strings"
 	"sync/atomic"
 	"testing"
@@ -51,19 +52,37 @@ func acquireTestLink(t *testing.T, d *linkLifecycleDriver) *model.Link {
 }
 
 func TestLinkLifecycleModes(t *testing.T) {
-	t.Run("TTL descriptor remains reusable after close", func(t *testing.T) {
+	t.Run("TTL borrowers are isolated while the descriptor is reused", func(t *testing.T) {
 		resetLinkLifecycleState(t)
 		ttl := time.Minute
 		d := &linkLifecycleDriver{
 			Storage: model.Storage{MountPath: "/ttl"},
-			links:   func() *model.Link { return &model.Link{URL: "https://example.test/file", Expiration: &ttl} },
+			links: func() *model.Link {
+				return &model.Link{
+					URL:        "https://example.test/file",
+					Header:     http.Header{"X-Link": {"cached"}},
+					Expiration: &ttl,
+				}
+			},
 		}
 
 		first := acquireTestLink(t, d)
+		if first.Expiration == nil || *first.Expiration != ttl {
+			t.Fatalf("borrower expiration = %v, want %v", first.Expiration, ttl)
+		}
+		*first.Expiration = time.Second
+		first.URL = "https://borrower.test/file"
+		first.Header.Set("X-Link", "borrower")
 		_ = first.Close()
 		second := acquireTestLink(t, d)
-		if second.URL != first.URL || d.calls.Load() != 1 {
-			t.Fatalf("TTL link was not reused: calls=%d", d.calls.Load())
+		if second.URL != "https://example.test/file" || second.Header.Get("X-Link") != "cached" {
+			t.Fatalf("cached descriptor was mutated: URL=%q Header=%q", second.URL, second.Header.Get("X-Link"))
+		}
+		if d.calls.Load() != 1 {
+			t.Fatalf("driver calls = %d, want 1", d.calls.Load())
+		}
+		if second.Expiration == nil || *second.Expiration != ttl {
+			t.Fatalf("cached expiration was mutated: %v", second.Expiration)
 		}
 		_ = second.Close()
 	})