Optimize STRM metadata sync

Co-authored-by: Cursor <cursoragent@cursor.com>

internal/strm/metadata.go

@@ -2,31 +2,40 @@ package strm
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"io"
+	"io/fs"
 	"net/http"
 	"os"
 	"path/filepath"
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"litepan/internal/playback"
 )
 
 const (
-	metadataHTTPAttempts  = 3
+	metadataCDNConcurrency = 3
-	metadataResolveRounds = 3
+	metadataHTTPAttempts   = 3
-	metadataClientTimeout = 5 * time.Minute
+	metadataResolveRounds  = 3
-	metadataFallbackUA    = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
+	metadataClientTimeout  = 5 * time.Minute
+	metadataFallbackUA     = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36"
 )
 
+type metadataResolver interface {
+	Resolve(ctx context.Context, accountID int64, fileID, ua string, refresh bool) (playback.Resolved, error)
+}
+
 type metadataSyncer struct {
-	playback   *playback.Service
+	playback   metadataResolver
 	failures   *FailureCollector
 	client     *http.Client
 	onProgress ScanProgressReporter
+	resolveMu  sync.Mutex
 }
 
 type metadataItem struct {
@@ -38,6 +47,16 @@ type metadataItem struct {
 	direct        bool
 }
 
+type metadataGroup struct {
+	fileID string
+	items  []metadataItem
+}
+
+type metadataDownloadJob struct {
+	group    metadataGroup
+	resolved playback.Resolved
+}
+
 func (m *metadataSyncer) syncFiles(ctx context.Context, accountID int64, root string, items []metadataItem) (int64, error) {
 	if m == nil || m.playback == nil || len(items) == 0 {
 		return 0, nil
@@ -52,23 +71,113 @@ func (m *metadataSyncer) syncFiles(ctx context.Context, accountID int64, root st
 	if client == nil {
 		client = &http.Client{Timeout: metadataClientTimeout}
 	}
+
+	var stateMu sync.Mutex
 	var created int64
-	for i, item := range pending {
+	done := 0
-		if err := ctx.Err(); err != nil {
+	finish := func(item metadataItem, made bool) {
-			return created, err
+		stateMu.Lock()
-		}
+		if made {
-		label := metadataProgressLabel(item.relPath)
-		reportMetadataProgress(m.onProgress, i, total, label)
-		ok, err := m.syncOne(ctx, client, accountID, root, item)
-		if err != nil {
-			return created, err
-		}
-		if ok {
 			created++
 		}
-		reportMetadataProgress(m.onProgress, i+1, total, label)
+		done++
+		reportMetadataProgress(m.onProgress, done, total, metadataProgressLabel(item.relPath))
+		stateMu.Unlock()
 	}
-	return created, nil
+	createdCount := func() int64 {
+		stateMu.Lock()
+		defer stateMu.Unlock()
+		return created
+	}
+
+	groups := make([]metadataGroup, 0, len(pending))
+	groupIndex := make(map[string]int, len(pending))
+	for _, item := range pending {
+		if err := ctx.Err(); err != nil {
+			return createdCount(), err
+		}
+		migrated, err := migrateLegacyMetadata(root, item)
+		if err != nil {
+			m.recordFailure(item.relPath, err.Error())
+			finish(item, false)
+			continue
+		}
+		if migrated {
+			finish(item, true)
+			continue
+		}
+		key := item.fileID
+		if key == "" {
+			key = "\x00" + item.relPath
+		}
+		if index, ok := groupIndex[key]; ok {
+			groups[index].items = append(groups[index].items, item)
+			continue
+		}
+		groupIndex[key] = len(groups)
+		groups = append(groups, metadataGroup{fileID: item.fileID, items: []metadataItem{item}})
+	}
+
+	jobs := make(chan metadataDownloadJob, metadataCDNConcurrency*2)
+	var workers sync.WaitGroup
+	workers.Add(metadataCDNConcurrency)
+	for range metadataCDNConcurrency {
+		go func() {
+			defer workers.Done()
+			for job := range jobs {
+				body, err := m.downloadResolvedWithRetry(ctx, client, accountID, job.group.fileID, job.resolved)
+				if err != nil {
+					if ctx.Err() == nil {
+						for _, item := range job.group.items {
+							m.recordFailure(item.relPath, err.Error())
+							finish(item, false)
+						}
+					}
+					continue
+				}
+				for _, item := range job.group.items {
+					if ctx.Err() != nil {
+						break
+					}
+					written, writeErr := writeMetadataFile(root, item.relPath, body)
+					if writeErr != nil {
+						m.recordFailure(item.relPath, writeErr.Error())
+						finish(item, false)
+						continue
+					}
+					finish(item, written)
+				}
+			}
+		}()
+	}
+
+	for _, group := range groups {
+		if err := ctx.Err(); err != nil {
+			break
+		}
+		resolved, err := m.resolve(ctx, accountID, group.fileID, false)
+		if err != nil {
+			if ctx.Err() != nil {
+				break
+			}
+			for _, item := range group.items {
+				m.recordFailure(item.relPath, err.Error())
+				finish(item, false)
+			}
+			continue
+		}
+		select {
+		case jobs <- metadataDownloadJob{group: group, resolved: resolved}:
+		case <-ctx.Done():
+			break
+		}
+	}
+	close(jobs)
+	workers.Wait()
+	if err := ctx.Err(); err != nil {
+		return createdCount(), err
+	}
+	return createdCount(), nil
 }
 
 func filterPendingMetadataItems(root string, items []metadataItem) []metadataItem {
@@ -103,77 +212,152 @@ func (m *metadataSyncer) syncOne(ctx context.Context, client *http.Client, accou
 	if info, statErr := os.Stat(dest); statErr == nil && info.Size() > 0 {
 		return false, nil
 	}
-	if item.legacyRelPath != "" {
+	if migrated, migrateErr := migrateLegacyMetadata(root, item); migrateErr != nil {
-		legacy := filepath.Join(root, item.legacyRelPath)
+		m.recordFailure(item.relPath, migrateErr.Error())
-		if info, statErr := os.Stat(legacy); statErr == nil && info.Size() > 0 {
+		return false, nil
-			if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+	} else if migrated {
-				m.recordFailure(item.relPath, err.Error())
+		return true, nil
-				return false, nil
-			}
-			if err := os.Rename(legacy, dest); err != nil {
-				m.recordFailure(item.relPath, err.Error())
-				return false, nil
-			}
-			return true, nil
-		}
 	}
 	body, dlErr := m.downloadWithRetry(ctx, client, accountID, item.fileID, 0)
 	if dlErr != nil {
 		m.recordFailure(item.relPath, dlErr.Error())
 		return false, nil
 	}
-	if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+	written, writeErr := writeMetadataFile(root, item.relPath, body)
-		m.recordFailure(item.relPath, err.Error())
+	if writeErr != nil {
+		m.recordFailure(item.relPath, writeErr.Error())
 		return false, nil
 	}
-	if err := os.WriteFile(dest, body, 0o644); err != nil {
+	return written, nil
-		m.recordFailure(item.relPath, err.Error())
+}
+
+func migrateLegacyMetadata(root string, item metadataItem) (bool, error) {
+	if item.legacyRelPath == "" {
 		return false, nil
 	}
+	legacy := filepath.Join(root, item.legacyRelPath)
+	if info, err := os.Stat(legacy); err != nil || info.Size() <= 0 {
+		return false, nil
+	}
+	dest := filepath.Join(root, item.relPath)
+	if _, err := os.Stat(dest); err == nil {
+		return false, nil
+	}
+	if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil {
+		return false, err
+	}
+	if err := os.Link(legacy, dest); err != nil {
+		if errors.Is(err, fs.ErrExist) {
+			return false, nil
+		}
+		return false, err
+	}
+	if err := os.Remove(legacy); err != nil {
+		return false, err
+	}
 	return true, nil
 }
 
+func writeMetadataFile(root, relPath string, body []byte) (bool, error) {
+	dest := filepath.Join(root, relPath)
+	if _, err := os.Stat(dest); err == nil {
+		return false, nil
+	}
+	dir := filepath.Dir(dest)
+	if err := os.MkdirAll(dir, 0o755); err != nil {
+		return false, err
+	}
+	tmp, err := os.CreateTemp(dir, "."+filepath.Base(dest)+".tmp-*")
+	if err != nil {
+		return false, err
+	}
+	tmpPath := tmp.Name()
+	defer os.Remove(tmpPath)
+	if err := tmp.Chmod(0o644); err != nil {
+		_ = tmp.Close()
+		return false, err
+	}
+	if _, err := tmp.Write(body); err != nil {
+		_ = tmp.Close()
+		return false, err
+	}
+	if err := tmp.Close(); err != nil {
+		return false, err
+	}
+	if err := os.Link(tmpPath, dest); err == nil {
+		return true, nil
+	} else if errors.Is(err, fs.ErrExist) {
+		return false, nil
+	} else {
+		return false, err
+	}
+}
+
+func (m *metadataSyncer) downloadResolvedWithRetry(
+	ctx context.Context,
+	client *http.Client,
+	accountID int64,
+	fileID string,
+	resolved playback.Resolved,
+) ([]byte, error) {
+	var lastErr error
+	res := resolved
+	for round := 0; round < metadataResolveRounds; round++ {
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		if round > 0 {
+			refreshed, err := m.resolve(ctx, accountID, fileID, true)
+			if err != nil {
+				lastErr = err
+				continue
+			}
+			res = refreshed
+		}
+		if res.Link.URL == "" {
+			lastErr = fmt.Errorf("无下载地址")
+			continue
+		}
+		size := res.File.Size
+		if size <= 0 && res.Link.Size > 0 {
+			size = res.Link.Size
+		}
+		body, err := fetchMetadataURLWithRetry(ctx, client, res.Link.URL, res.Link.Headers, size)
+		if err == nil {
+			return body, nil
+		}
+		lastErr = err
+	}
+	if lastErr == nil {
+		lastErr = fmt.Errorf("元数据下载失败")
+	}
+	return nil, lastErr
+}
+
+func (m *metadataSyncer) downloadWithRetry(ctx context.Context, client *http.Client, accountID int64, fileID string, expectedSize int64) ([]byte, error) {
+	res, err := m.resolve(ctx, accountID, fileID, false)
+	if err != nil {
+		return nil, err
+	}
+	if expectedSize > 0 {
+		res.File.Size = expectedSize
+		res.Link.Size = expectedSize
+	}
+	return m.downloadResolvedWithRetry(ctx, client, accountID, fileID, res)
+}
+
+func (m *metadataSyncer) resolve(ctx context.Context, accountID int64, fileID string, refresh bool) (playback.Resolved, error) {
+	m.resolveMu.Lock()
+	defer m.resolveMu.Unlock()
+	return m.playback.Resolve(ctx, accountID, fileID, "", refresh)
+}
+
 func (m *metadataSyncer) recordFailure(path, reason string) {
 	if m.failures != nil {
 		m.failures.Add(ScanFailureMetadata, path, reason)
 	}
 }
 
-func (m *metadataSyncer) downloadWithRetry(ctx context.Context, client *http.Client, accountID int64, fileID string, expectedSize int64) ([]byte, error) {
-	var lastErr error
-	for round := 0; round < metadataResolveRounds; round++ {
-		if err := ctx.Err(); err != nil {
-			return nil, err
-		}
-		refresh := round > 0
-		res, err := m.playback.Resolve(ctx, accountID, fileID, "", refresh)
-		if err != nil {
-			lastErr = err
-			continue
-		}
-		if res.Link.URL == "" {
-			lastErr = fmt.Errorf("无下载地址")
-			continue
-		}
-		size := expectedSize
-		if size <= 0 && res.File.Size > 0 {
-			size = res.File.Size
-		}
-		body, fetchErr := fetchMetadataURLWithRetry(ctx, client, res.Link.URL, res.Link.Headers, size)
-		if fetchErr == nil {
-			return body, nil
-		}
-		lastErr = fetchErr
-		if isIntegrityMetadataErr(fetchErr) || isRefreshMetadataErr(fetchErr) {
-			continue
-		}
-	}
-	if lastErr == nil {
-		lastErr = fmt.Errorf("元数据下载失败")
-	}
-	return nil, lastErr
-}
-
 func fetchMetadataURLWithRetry(ctx context.Context, client *http.Client, downloadURL string, headers http.Header, expectedSize int64) ([]byte, error) {
 	var lastErr error
 	for attempt := 0; attempt < metadataHTTPAttempts; attempt++ {
@@ -282,20 +466,6 @@ func isIntegrityMetadataErr(err error) bool {
 	return strings.Contains(err.Error(), "不一致")
 }
 
-func isRefreshMetadataErr(err error) bool {
-	if err == nil {
-		return false
-	}
-	msg := strings.ToLower(err.Error())
-	if strings.Contains(msg, "invalid signature") {
-		return true
-	}
-	if strings.Contains(msg, "http 403") || strings.Contains(msg, "http 401") {
-		return true
-	}
-	return false
-}
-
 func metadataRelPath(outputFolder string, relDirs []string, fileName string) string {
 	parts := make([]string, 0, len(relDirs)+2)
 	parts = append(parts, SafeName(outputFolder))

internal/strm/metadata_test.go

@@ -0,0 +1,299 @@
+package strm
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"litepan/internal/domain"
+	"litepan/internal/playback"
+)
+
+type metadataResolverStub struct {
+	mu           sync.Mutex
+	baseURL      string
+	calls        map[string]int
+	active       int
+	maxActive    int
+	resolveDelay time.Duration
+}
+
+func (s *metadataResolverStub) Resolve(
+	ctx context.Context,
+	_ int64,
+	fileID, _ string,
+	_ bool,
+) (playback.Resolved, error) {
+	s.mu.Lock()
+	if s.calls == nil {
+		s.calls = make(map[string]int)
+	}
+	s.calls[fileID]++
+	s.active++
+	if s.active > s.maxActive {
+		s.maxActive = s.active
+	}
+	s.mu.Unlock()
+
+	defer func() {
+		s.mu.Lock()
+		s.active--
+		s.mu.Unlock()
+	}()
+	if s.resolveDelay > 0 {
+		timer := time.NewTimer(s.resolveDelay)
+		select {
+		case <-ctx.Done():
+			timer.Stop()
+			return playback.Resolved{}, ctx.Err()
+		case <-timer.C:
+		}
+	}
+	return playback.Resolved{
+		Link: domain.DownloadInfo{URL: s.baseURL + "/" + fileID},
+	}, nil
+}
+
+func (s *metadataResolverStub) callCount(fileID string) int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.calls[fileID]
+}
+
+func (s *metadataResolverStub) peakResolveConcurrency() int {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	return s.maxActive
+}
+
+func TestMetadataSyncerPipelinesSerialResolveAndThreeCDNDownloads(t *testing.T) {
+	var active atomic.Int32
+	var peak atomic.Int32
+	entered := make(chan struct{}, 8)
+	release := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		current := active.Add(1)
+		defer active.Add(-1)
+		for {
+			previous := peak.Load()
+			if current <= previous || peak.CompareAndSwap(previous, current) {
+				break
+			}
+		}
+		entered <- struct{}{}
+		select {
+		case <-release:
+			_, _ = w.Write([]byte(filepath.Base(r.URL.Path)))
+		case <-r.Context().Done():
+		}
+	}))
+	defer server.Close()
+
+	resolver := &metadataResolverStub{baseURL: server.URL, resolveDelay: 5 * time.Millisecond}
+	items := make([]metadataItem, 0, 6)
+	for i := range 6 {
+		id := string(rune('a' + i))
+		items = append(items, metadataItem{fileID: id, relPath: id + ".nfo"})
+	}
+	root := t.TempDir()
+	type result struct {
+		created int64
+		err     error
+	}
+	done := make(chan result, 1)
+	go func() {
+		created, err := (&metadataSyncer{playback: resolver}).syncFiles(t.Context(), 1, root, items)
+		done <- result{created: created, err: err}
+	}()
+
+	for range metadataCDNConcurrency {
+		select {
+		case <-entered:
+		case <-time.After(2 * time.Second):
+			t.Fatal("未形成 3 路 CDN 并发")
+		}
+	}
+	select {
+	case <-entered:
+		t.Fatal("CDN 下载并发超过 3")
+	case <-time.After(100 * time.Millisecond):
+	}
+	close(release)
+
+	got := <-done
+	if got.err != nil {
+		t.Fatal(got.err)
+	}
+	if got.created != int64(len(items)) {
+		t.Fatalf("新增数=%d，期望=%d", got.created, len(items))
+	}
+	if peak.Load() != metadataCDNConcurrency {
+		t.Fatalf("CDN 峰值并发=%d，期望=%d", peak.Load(), metadataCDNConcurrency)
+	}
+	if resolver.peakResolveConcurrency() != 1 {
+		t.Fatalf("取直链峰值并发=%d，期望=1", resolver.peakResolveConcurrency())
+	}
+}
+
+func TestMetadataSyncerDeduplicatesFileIDAndPreservesExistingFiles(t *testing.T) {
+	var downloads atomic.Int32
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		downloads.Add(1)
+		_, _ = w.Write([]byte("metadata"))
+	}))
+	defer server.Close()
+
+	root := t.TempDir()
+	existing := filepath.Join(root, "existing.nfo")
+	if err := os.WriteFile(existing, []byte("scraped"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	resolver := &metadataResolverStub{baseURL: server.URL}
+	items := []metadataItem{
+		{fileID: "same", relPath: "one.nfo"},
+		{fileID: "same", relPath: "two.nfo"},
+		{fileID: "existing", relPath: "existing.nfo"},
+	}
+	created, err := (&metadataSyncer{playback: resolver}).syncFiles(t.Context(), 1, root, items)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if created != 2 {
+		t.Fatalf("新增数=%d，期望=2", created)
+	}
+	if resolver.callCount("same") != 1 || resolver.callCount("existing") != 0 {
+		t.Fatalf("取直链次数 same=%d existing=%d", resolver.callCount("same"), resolver.callCount("existing"))
+	}
+	if downloads.Load() != 1 {
+		t.Fatalf("CDN 下载次数=%d，期望=1", downloads.Load())
+	}
+	if body, err := os.ReadFile(existing); err != nil || string(body) != "scraped" {
+		t.Fatalf("已有刮削文件被改写：%q, err=%v", body, err)
+	}
+}
+
+func TestMetadataSyncerSerializesRefreshResolve(t *testing.T) {
+	var requestMu sync.Mutex
+	requests := make(map[string]int)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requestMu.Lock()
+		requests[r.URL.Path]++
+		count := requests[r.URL.Path]
+		requestMu.Unlock()
+		if count == 1 {
+			http.Error(w, "expired", http.StatusForbidden)
+			return
+		}
+		_, _ = w.Write([]byte("metadata"))
+	}))
+	defer server.Close()
+
+	resolver := &metadataResolverStub{baseURL: server.URL, resolveDelay: 10 * time.Millisecond}
+	items := []metadataItem{
+		{fileID: "one", relPath: "one.nfo"},
+		{fileID: "two", relPath: "two.nfo"},
+		{fileID: "three", relPath: "three.nfo"},
+	}
+	created, err := (&metadataSyncer{playback: resolver}).syncFiles(t.Context(), 1, t.TempDir(), items)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if created != int64(len(items)) {
+		t.Fatalf("新增数=%d，期望=%d", created, len(items))
+	}
+	if resolver.peakResolveConcurrency() != 1 {
+		t.Fatalf("刷新取直链峰值并发=%d，期望=1", resolver.peakResolveConcurrency())
+	}
+	for _, item := range items {
+		if got := resolver.callCount(item.fileID); got != 2 {
+			t.Fatalf("%s 取直链次数=%d，期望首次+刷新共2次", item.fileID, got)
+		}
+	}
+}
+
+func TestMetadataSyncerDoesNotOverwriteFileCreatedDuringDownload(t *testing.T) {
+	started := make(chan struct{}, 1)
+	release := make(chan struct{})
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+		started <- struct{}{}
+		<-release
+		_, _ = w.Write([]byte("remote"))
+	}))
+	defer server.Close()
+
+	root := t.TempDir()
+	target := filepath.Join(root, "movie.nfo")
+	resolver := &metadataResolverStub{baseURL: server.URL}
+	done := make(chan struct {
+		created int64
+		err     error
+	}, 1)
+	go func() {
+		created, err := (&metadataSyncer{playback: resolver}).syncFiles(t.Context(), 1, root, []metadataItem{
+			{fileID: "movie", relPath: "movie.nfo"},
+		})
+		done <- struct {
+			created int64
+			err     error
+		}{created: created, err: err}
+	}()
+	select {
+	case <-started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("元数据下载未开始")
+	}
+	if err := os.WriteFile(target, []byte("scraped"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	close(release)
+	got := <-done
+	if got.err != nil {
+		t.Fatal(got.err)
+	}
+	if got.created != 0 {
+		t.Fatalf("新增数=%d，期望不覆盖并发生成的文件", got.created)
+	}
+	if body, err := os.ReadFile(target); err != nil || string(body) != "scraped" {
+		t.Fatalf("并发刮削文件被覆盖：%q, err=%v", body, err)
+	}
+}
+
+func TestMetadataSyncerCancellationStopsWorkers(t *testing.T) {
+	started := make(chan struct{}, 1)
+	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		started <- struct{}{}
+		<-r.Context().Done()
+	}))
+	defer server.Close()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	resolver := &metadataResolverStub{baseURL: server.URL}
+	done := make(chan error, 1)
+	go func() {
+		_, err := (&metadataSyncer{playback: resolver}).syncFiles(ctx, 1, t.TempDir(), []metadataItem{
+			{fileID: "cancel", relPath: "cancel.nfo"},
+		})
+		done <- err
+	}()
+	select {
+	case <-started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("元数据下载未开始")
+	}
+	cancel()
+	select {
+	case err := <-done:
+		if !errors.Is(err, context.Canceled) {
+			t.Fatalf("取消后错误=%v，期望 context.Canceled", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("取消后下载协程未退出")
+	}
+}