From 2c65400a39b132fc7cf0e39f9b9f0f47a90cd6e7 Mon Sep 17 00:00:00 2001 From: King Date: Mon, 27 Jul 2026 23:16:57 +0800 Subject: [PATCH] Optimize STRM metadata sync Co-authored-by: Cursor --- internal/strm/metadata.go | 338 +++++++++++++++++++++++++-------- internal/strm/metadata_test.go | 299 +++++++++++++++++++++++++++++ 2 files changed, 553 insertions(+), 84 deletions(-) create mode 100644 internal/strm/metadata_test.go diff --git a/internal/strm/metadata.go b/internal/strm/metadata.go index 9e084f8..3d4d974 100644 --- a/internal/strm/metadata.go +++ b/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 - metadataResolveRounds = 3 - 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" + metadataCDNConcurrency = 3 + metadataHTTPAttempts = 3 + metadataResolveRounds = 3 + 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 { - if err := ctx.Err(); err != nil { - return created, err - } - 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 { + done := 0 + finish := func(item metadataItem, made bool) { + stateMu.Lock() + if made { 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 != "" { - legacy := filepath.Join(root, item.legacyRelPath) - if info, statErr := os.Stat(legacy); statErr == nil && info.Size() > 0 { - if err := os.MkdirAll(filepath.Dir(dest), 0o755); err != nil { - m.recordFailure(item.relPath, err.Error()) - return false, nil - } - if err := os.Rename(legacy, dest); err != nil { - m.recordFailure(item.relPath, err.Error()) - return false, nil - } - return true, nil - } + if migrated, migrateErr := migrateLegacyMetadata(root, item); migrateErr != nil { + m.recordFailure(item.relPath, migrateErr.Error()) + return false, nil + } else if migrated { + 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 { - m.recordFailure(item.relPath, err.Error()) + written, writeErr := writeMetadataFile(root, item.relPath, body) + if writeErr != nil { + m.recordFailure(item.relPath, writeErr.Error()) return false, nil } - if err := os.WriteFile(dest, body, 0o644); err != nil { - m.recordFailure(item.relPath, err.Error()) + return written, nil +} + +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)) diff --git a/internal/strm/metadata_test.go b/internal/strm/metadata_test.go new file mode 100644 index 0000000..848e1dc --- /dev/null +++ b/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("取消后下载协程未退出") + } +}