diff --git a/internal/model/strm.go b/internal/model/strm.go index 4799c8d..4f02b1a 100644 --- a/internal/model/strm.go +++ b/internal/model/strm.go @@ -128,7 +128,7 @@ type StrmUploadTask struct { FileName string `gorm:"size:512" json:"file_name"` LocalPath string `gorm:"size:1024" json:"local_path"` // 本地源文件 RemotePath string `gorm:"size:1024" json:"remote_path"` // 远端目标路径 - RemoteRef string `gorm:"size:1024" json:"remote_ref"` // 远端同名旧文件引用（115 文件 ID；上传覆盖前先删除旧文件，WebDAV/OpenList 直接覆盖无需删除） + RemoteRef string `gorm:"size:1024" json:"remote_ref"` // 上传前：远端同名旧文件 ID（逗号分隔，覆盖前先删）；上传成功后：新文件 ID Size int64 `json:"size"` Status string `gorm:"size:16;index" json:"status"` Error string `gorm:"size:1024" json:"error"` diff --git a/internal/repository/strm_repository.go b/internal/repository/strm_repository.go index 7b1f2d7..6d0daf1 100644 --- a/internal/repository/strm_repository.go +++ b/internal/repository/strm_repository.go @@ -861,6 +861,34 @@ func (r *StrmUploadTaskRepository) GetActiveLocalPathMap(ctx context.Context, sy return out, nil } +// GetRecentDoneUploadSizeMap 返回近期已成功上传的 local_path → size。 +// 用于缩短「上传已 done 但 115 列表尚未反映」窗口内的重复入队：同路径且大小未变则跳过。 +// 同一路径存在多条 done 时取最新一条（finished_at 降序）。 +func (r *StrmUploadTaskRepository) GetRecentDoneUploadSizeMap(ctx context.Context, syncPathID string, since time.Time) (map[string]int64, error) { + var rows []model.StrmUploadTask + err := r.db.WithContext(ctx).Model(&model.StrmUploadTask{}). + Select("local_path", "size", "finished_at"). + Where("sync_path_id = ? AND status = ? AND finished_at IS NOT NULL AND finished_at >= ?", + syncPathID, model.StrmTaskDone, since). + Order("finished_at DESC"). + Find(&rows).Error + if err != nil { + return nil, err + } + out := make(map[string]int64, len(rows)) + for _, row := range rows { + if row.LocalPath == "" { + continue + } + // 已按 finished_at DESC；先写入的是最新，后续同路径跳过 + if _, exists := out[row.LocalPath]; exists { + continue + } + out[row.LocalPath] = row.Size + } + return out, nil +} + func (r *StrmUploadTaskRepository) DeleteFinishedOlderThan(ctx context.Context, before time.Time) error { return withSQLiteBusyRetry(ctx, func() error { return r.db.WithContext(ctx).Unscoped().Where("status IN ? AND finished_at < ?", diff --git a/internal/service/cloud115/open.go b/internal/service/cloud115/open.go index f0887c5..0769ad1 100644 --- a/internal/service/cloud115/open.go +++ b/internal/service/cloud115/open.go @@ -119,6 +119,60 @@ func (c *OpenClient) GetFsListFlat(ctx context.Context, cid string, offset, limi return files, resp.Count, nil } +// FindNamedContentInParent 在父目录下查找与本地内容一致的同名文件。 +// 匹配条件：文件名完全一致、大小一致，且 SHA1 为空或与 expectedSHA1 大小写不敏感相等。 +// 同时返回该目录下全部同名文件（含未匹配的脏副本），便于上传前清理。 +// 目录过大时分页扫描，最多拉取 maxListPages 页（每页 pageSize 条）。 +func (c *OpenClient) FindNamedContentInParent(ctx context.Context, parentCID, fileName, expectedSHA1 string, expectedSize int64) (matched *RemoteFile, sameName []RemoteFile, err error) { + fileName = strings.TrimSpace(fileName) + if fileName == "" { + return nil, nil, nil + } + const pageSize = 200 + const maxListPages = 20 // 最多扫描 4000 项，元数据父目录通常远小于此 + expectedSHA1 = strings.TrimSpace(expectedSHA1) + offset := 0 + for page := 0; page < maxListPages; page++ { + files, _, listErr := c.GetFsList(ctx, parentCID, offset, pageSize) + if listErr != nil { + return nil, sameName, listErr + } + if len(files) == 0 { + break + } + for i := range files { + f := files[i] + if f.Category == TypeDir { + continue + } + // fta=0/2 表示未上传完成，不可作为已存在副本 + if f.Fta == "0" || f.Fta == "2" { + continue + } + if f.FileName != fileName { + continue + } + sameName = append(sameName, f) + if matched != nil { + continue + } + if f.FileSize != expectedSize { + continue + } + remoteSha := strings.TrimSpace(f.Sha1) + if remoteSha == "" || remoteSha == "-" || strings.EqualFold(remoteSha, expectedSHA1) { + cp := f + matched = &cp + } + } + if len(files) < pageSize { + break + } + offset += len(files) + } + return matched, sameName, nil +} + // GetFsDetailByCid 查询文件（夹）详情。 func (c *OpenClient) GetFsDetailByCid(ctx context.Context, fileId string) (*RemoteFileDetail, error) { params := map[string]string{"file_id": fileId} diff --git a/internal/service/cloud115/open_find_test.go b/internal/service/cloud115/open_find_test.go new file mode 100644 index 0000000..24f2a2d --- /dev/null +++ b/internal/service/cloud115/open_find_test.go @@ -0,0 +1,54 @@ +package cloud115 + +import ( + "context" + "fmt" + "net/http" + "testing" +) + +func TestFindNamedContentInParentMatchAndSameName(t *testing.T) { + sha := "AABBCCDDEEFF00112233445566778899AABBCCDD" + mockAPI(t, func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/open/ufile/files" { + t.Errorf("unexpected path %s", r.URL.Path) + } + w.Write([]byte(fmt.Sprintf(`{"state":true,"data":[ + {"fid":"dir-1","fc":"0","fn":"sub","fs":0}, + {"fid":"keep","fc":"1","fn":"a.nfo","fs":10,"sha1":%q,"fta":"1"}, + {"fid":"dirty","fc":"1","fn":"a.nfo","fs":10,"sha1":"OTHER","fta":"1"}, + {"fid":"other","fc":"1","fn":"b.nfo","fs":10,"sha1":%q,"fta":"1"}, + {"fid":"incomplete","fc":"1","fn":"a.nfo","fs":10,"sha1":%q,"fta":"0"} + ]}`, sha, sha, sha))) + }) + c := NewOpenClient("100195125", "at1", "rt1") + matched, sameName, err := c.FindNamedContentInParent(context.Background(), "parent", "a.nfo", sha, 10) + if err != nil { + t.Fatal(err) + } + if matched == nil || matched.FileId != "keep" { + t.Fatalf("matched = %+v, want keep", matched) + } + if len(sameName) != 2 { // keep + dirty；incomplete 被 fta 过滤 + t.Fatalf("sameName len=%d, want 2 (incomplete excluded)", len(sameName)) + } +} + +func TestFindNamedContentInParentNoMatch(t *testing.T) { + mockAPI(t, func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte(`{"state":true,"data":[ + {"fid":"x","fc":"1","fn":"a.nfo","fs":9,"sha1":"OTHER","fta":"1"} + ]}`)) + }) + c := NewOpenClient("100195125", "at1", "rt1") + matched, sameName, err := c.FindNamedContentInParent(context.Background(), "parent", "a.nfo", "WANT", 10) + if err != nil { + t.Fatal(err) + } + if matched != nil { + t.Fatalf("expected no match, got %+v", matched) + } + if len(sameName) != 1 { + t.Fatalf("sameName should still list name hits, got %d", len(sameName)) + } +} diff --git a/internal/service/strm_queue.go b/internal/service/strm_queue.go index 3b63c52..e1651a7 100644 --- a/internal/service/strm_queue.go +++ b/internal/service/strm_queue.go @@ -26,6 +26,8 @@ import ( const ( strmMaxTaskRetry = 3 + // strmRecentUploadSkipWindow：上传已成功但 115 列表尚未反映时，同步扫描跳过同路径同大小再入队的宽限窗口。 + strmRecentUploadSkipWindow = 30 * time.Minute ) // downloadWorker 下载队列 worker：认领 → 解析直链 → 下载 → 落盘。 @@ -332,18 +334,28 @@ func (s *StrmService) processUploadTask(ctx context.Context, task *model.StrmUpl } // processUpload115 115 元数据上传：task.RemotePath 存的是父目录 cid，FileName 为远端文件名。 +// +// 幂等要点： +// 1. 上传/重试前按父目录 + 文件名 + SHA1 探活：远端已有同内容副本则跳过上传，仅清理其它脏副本； +// 2. 真正上传成功后把新 file_id 写回 RemoteRef，供下次同步/重试识别； +// 3. 115 上传不保证同名覆盖，内容不同时仍先删旧再传。 func (s *StrmService) processUpload115(ctx context.Context, task *model.StrmUploadTask) { - finish := func(status, message string) { + finish := func(status, message, remoteRef string) { now := time.Now() task.Status = status task.Error = message task.FinishedAt = &now - // 条件化收尾：与下载侧一致，防止覆盖已取消任务。 - if ok, err := s.repo.StrmUpload.UpdateIfRunning(context.Background(), task.ID, map[string]any{ + updates := map[string]any{ "status": status, "error": message, "finished_at": &now, - }); err != nil { + } + if remoteRef != "" { + task.RemoteRef = remoteRef + updates["remote_ref"] = remoteRef + } + // 条件化收尾：与下载侧一致，防止覆盖已取消任务。 + if ok, err := s.repo.StrmUpload.UpdateIfRunning(context.Background(), task.ID, updates); err != nil { s.log.Warn("update strm upload task failed", zap.Error(err)) } else if !ok { s.log.Info("strm upload task already closed elsewhere", zap.String("id", task.ID)) @@ -351,7 +363,7 @@ func (s *StrmService) processUpload115(ctx context.Context, task *model.StrmUplo } acct, err := s.repo.StrmAccount.FindByID(ctx, task.AccountID) if err != nil || acct == nil { - finish(model.StrmTaskFailed, "网盘账号不存在") + finish(model.StrmTaskFailed, "网盘账号不存在", "") return } provider, err := s.providerFor(ctx, acct) @@ -359,58 +371,98 @@ func (s *StrmService) processUpload115(ctx context.Context, task *model.StrmUplo s.uploadTaskFailWithRetry(task, err.Error()) return } - named, ok := provider.(interface { - PutFileNamed(ctx context.Context, parentCID, fileName string, r io.Reader) error - }) + open115, ok := provider.(cloud.OpenAPI115Provider) if !ok { - finish(model.StrmTaskFailed, "该网盘不支持元数据上传") + finish(model.StrmTaskFailed, "该网盘不支持元数据上传", "") return } - // 以本地为准：网盘端已有同名但内容不同的旧元数据时，先尝试批量删除所有旧副本再上传。 - // 115 的上传接口不保证同名覆盖，直接上传可能产生同名重复文件。 - // 删除失败时不中止任务——继续上传新文件，旧副本交由下次同步的 cleanupBatchRedundantFiles - // 按目录批量清理（下次同步会看到新旧两个版本，命中新版本后把旧版本 cid 收入 pendingDeletes - // 异步删除）。这样避免了「删旧失败 → 任务重试 → 再次删旧失败 → 永远无法上传」的死循环。 - if task.RemoteRef != "" { - open115, ok := provider.(cloud.OpenAPI115Provider) - if !ok { - finish(model.StrmTaskFailed, "该网盘不支持删除远端旧元数据") - return - } - refs := strings.Split(task.RemoteRef, ",") - if err := open115.OpenClient().DeleteFiles(ctx, task.RemotePath, refs...); err != nil { - s.log.Warn("删除网盘旧元数据失败，跳过删除继续上传新文件", - zap.String("task_id", task.ID), - zap.String("local_path", task.LocalPath), - zap.Error(err)) - // 不 return：继续上传新文件，旧副本由下次同步清理 - } - } - // 优先使用直接本地文件上传接口，零拷贝且彻底根除并发临时文件同名碰撞 - if localUploader, ok := provider.(interface { - PutLocalFile(ctx context.Context, parentCID, localPath string) error - }); ok { - if err := localUploader.PutLocalFile(ctx, task.RemotePath, task.LocalPath); err != nil { - s.uploadTaskFailWithRetry(task, "上传失败："+err.Error()) - return - } - finish(model.StrmTaskDone, "") - return - } + client := open115.OpenClient() - f, err := os.Open(task.LocalPath) - if err != nil { - s.uploadTaskFailWithRetry(task, "打开本地文件失败："+err.Error()) - return - } - if err := named.PutFileNamed(ctx, task.RemotePath, task.FileName, f); err != nil { - _ = f.Close() - s.uploadTaskFailWithRetry(task, "上传失败："+err.Error()) - return - } - _ = f.Close() - finish(model.StrmTaskDone, "") + localSHA1, shaErr := cloud115.FileSHA1(task.LocalPath) + if shaErr != nil { + s.uploadTaskFailWithRetry(task, "计算本地 SHA1 失败："+shaErr.Error()) + return } + info, statErr := os.Stat(task.LocalPath) + if statErr != nil { + s.uploadTaskFailWithRetry(task, "读取本地文件失败："+statErr.Error()) + return + } + localSize := info.Size() + + // ── 探活：父目录下是否已有同名同内容副本（覆盖「上传成功但本地当失败重试」）── + matched, sameName, probeErr := client.FindNamedContentInParent(ctx, task.RemotePath, task.FileName, localSHA1, localSize) + if probeErr != nil { + // 探活失败不阻断上传：按原路径继续，避免列表接口抖动导致任务永久卡住 + s.log.Warn("115 上传前探活失败，继续上传", + zap.String("task_id", task.ID), + zap.String("local_path", task.LocalPath), + zap.Error(probeErr)) + } else if matched != nil && matched.FileId != "" { + staleIDs := collectStale115FileIDs(task.RemoteRef, sameName, matched.FileId) + if len(staleIDs) > 0 { + if err := client.DeleteFiles(ctx, task.RemotePath, staleIDs...); err != nil { + s.log.Warn("探活命中后清理 115 脏副本失败（已跳过上传）", + zap.String("task_id", task.ID), + zap.String("matched_id", matched.FileId), + zap.Error(err)) + } + } + s.log.Info("115 元数据已存在同内容副本，跳过上传", + zap.String("task_id", task.ID), + zap.String("local_path", task.LocalPath), + zap.String("file_id", matched.FileId)) + finish(model.StrmTaskDone, "", matched.FileId) + return + } + + // ── 需要上传：先尽量删掉任务携带的旧副本，再真正上传 ── + // 删除失败时不中止——继续上传新文件，旧副本交由下次同步 cleanupBatchRedundantFiles。 + if task.RemoteRef != "" { + refs := strings.Split(task.RemoteRef, ",") + if err := client.DeleteFiles(ctx, task.RemotePath, refs...); err != nil { + s.log.Warn("删除网盘旧元数据失败，跳过删除继续上传新文件", + zap.String("task_id", task.ID), + zap.String("local_path", task.LocalPath), + zap.Error(err)) + } + } + + result, err := client.Upload(ctx, task.LocalPath, task.RemotePath, "", "") + if err != nil { + s.uploadTaskFailWithRetry(task, "上传失败："+err.Error()) + return + } + newID := "" + if result != nil { + newID = strings.TrimSpace(result.FileId) + } + finish(model.StrmTaskDone, "", newID) +} + +// collectStale115FileIDs 汇总待删脏副本：任务 RemoteRef + 探活所见同名文件，排除 keepID。 +func collectStale115FileIDs(remoteRef string, sameName []cloud115.RemoteFile, keepID string) []string { + seen := map[string]struct{}{} + var out []string + add := func(id string) { + id = strings.TrimSpace(id) + if id == "" || id == keepID { + return + } + if _, ok := seen[id]; ok { + return + } + seen[id] = struct{}{} + out = append(out, id) + } + for _, id := range strings.Split(remoteRef, ",") { + add(id) + } + for _, f := range sameName { + add(f.FileId) + } + return out +} // downloadTaskFailWithRetry 下载失败任务按退避重试，超过上限标记 failed。 func (s *StrmService) downloadTaskFailWithRetry(task *model.StrmDownloadTask, message string) { diff --git a/internal/service/strm_queue_test.go b/internal/service/strm_queue_test.go index 2a3a1b8..f7643f1 100644 --- a/internal/service/strm_queue_test.go +++ b/internal/service/strm_queue_test.go @@ -3,10 +3,12 @@ package service import ( "context" "errors" + "fmt" "net/http" "net/http/httptest" "os" "path/filepath" + "strings" "sync" "testing" "time" @@ -174,8 +176,8 @@ func TestRequeueDownloadTask(t *testing.T) { } // TestProcessUpload115DeletesStaleRemoteMetaFirst 验证 115 覆盖上传语义（以本地为准）： -// 任务携带网盘旧文件 ID 时，必须先调用 /open/ufile/delete 删除旧元数据再上传本地文件， -// 避免 115 出现同名重复文件；删除请求应携带 file_ids 与父目录 cid。 +// 探活未命中时，任务携带网盘旧文件 ID 必须先 /open/ufile/delete 再上传； +// 上传成功后 RemoteRef 回写为新 file_id。 func TestProcessUpload115DeletesStaleRemoteMetaFirst(t *testing.T) { svc := testStrmService(t) localDir := t.TempDir() @@ -197,6 +199,10 @@ func TestProcessUpload115DeletesStaleRemoteMetaFirst(t *testing.T) { mu.Lock() defer mu.Unlock() switch r.URL.Path { + case "/open/ufile/files": + calls = append(calls, "probe") + // 探活：目录为空 / 无同内容副本 → 继续删旧上传 + w.Write([]byte(`{"state":true,"data":[]}`)) case "/open/ufile/delete": calls = append(calls, "delete") deleteForm["file_ids"] = r.FormValue("file_ids") @@ -232,10 +238,13 @@ func TestProcessUpload115DeletesStaleRemoteMetaFirst(t *testing.T) { if task.Status != model.StrmTaskDone { t.Fatalf("upload task should succeed, status = %s, error = %s", task.Status, task.Error) } + if task.RemoteRef != "new-1" { + t.Fatalf("successful upload should record new file_id in RemoteRef, got %q", task.RemoteRef) + } mu.Lock() defer mu.Unlock() - if len(calls) != 2 || calls[0] != "delete" || calls[1] != "upload" { - t.Fatalf("expected delete before upload, got calls = %v", calls) + if len(calls) != 3 || calls[0] != "probe" || calls[1] != "delete" || calls[2] != "upload" { + t.Fatalf("expected probe→delete→upload, got calls = %v", calls) } if deleteForm["file_ids"] != "old-file-1" { t.Fatalf("delete file_ids = %q, want old-file-1", deleteForm["file_ids"]) @@ -245,6 +254,97 @@ func TestProcessUpload115DeletesStaleRemoteMetaFirst(t *testing.T) { } } +// TestProcessUpload115SkipsWhenProbeFindsSameContent 验证上传/重试前探活命中同内容副本时跳过上传， +// 并清理其它脏副本，RemoteRef 回写为已存在的 file_id。 +func TestProcessUpload115SkipsWhenProbeFindsSameContent(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + localFile := filepath.Join(localDir, "movie.nfo") + content := []byte("already-on-115") + if err := os.WriteFile(localFile, content, 0o644); err != nil { + t.Fatal(err) + } + sha, err := cloud115.FileSHA1(localFile) + if err != nil { + t.Fatal(err) + } + + acct := &model.StrmAccount{Name: "fake115", Provider: "cloud115", Config: "{}", Enabled: true} + if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil { + t.Fatal(err) + } + + var mu sync.Mutex + var calls []string + deleteForm := map[string]string{} + api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _ = r.ParseForm() + mu.Lock() + defer mu.Unlock() + switch r.URL.Path { + case "/open/ufile/files": + calls = append(calls, "probe") + w.Write([]byte(fmt.Sprintf( + `{"state":true,"data":[ + {"fid":"keep-1","fc":"1","fn":"movie.nfo","fs":%d,"sha1":%q,"fta":"1"}, + {"fid":"dirty-2","fc":"1","fn":"movie.nfo","fs":9,"sha1":"DEADBEEF","fta":"1"} + ]}`, len(content), sha))) + case "/open/ufile/delete": + calls = append(calls, "delete") + deleteForm["file_ids"] = r.FormValue("file_ids") + w.Write([]byte(`{"state":true,"data":[]}`)) + case "/open/upload/init": + calls = append(calls, "upload") + w.Write([]byte(`{"state":true,"data":{"status":2,"file_id":"should-not","pick_code":"x","callback":null}}`)) + default: + t.Errorf("unexpected 115 api path %s", r.URL.Path) + w.Write([]byte(`{"state":false,"message":"unexpected path"}`)) + } + })) + defer api.Close() + oldPro := cloud115.ProAPIBase + cloud115.ProAPIBase = api.URL + defer func() { cloud115.ProAPIBase = oldPro }() + + task := &model.StrmUploadTask{ + Base: model.Base{ID: "up-skip-1"}, + SyncPathID: "p1", + AccountID: acct.ID, + Provider: model.StrmProvider115, + FileName: "movie.nfo", + LocalPath: localFile, + RemotePath: "777", + RemoteRef: "old-ref", + Size: int64(len(content)), + Status: model.StrmTaskRunning, + } + svc.processUpload115(context.Background(), task) + + if task.Status != model.StrmTaskDone { + t.Fatalf("probe hit should finish done, status=%s err=%s", task.Status, task.Error) + } + if task.RemoteRef != "keep-1" { + t.Fatalf("RemoteRef should be matched file_id keep-1, got %q", task.RemoteRef) + } + mu.Lock() + defer mu.Unlock() + if len(calls) != 2 || calls[0] != "probe" || calls[1] != "delete" { + t.Fatalf("expected probe→delete (no upload), got %v", calls) + } + // 脏副本 dirty-2 与任务旧 ref 都应被清理，keep-1 不得出现 + ids := strings.Split(deleteForm["file_ids"], ",") + idSet := map[string]bool{} + for _, id := range ids { + idSet[strings.TrimSpace(id)] = true + } + if !idSet["dirty-2"] || !idSet["old-ref"] { + t.Fatalf("delete should include dirty-2 and old-ref, got %q", deleteForm["file_ids"]) + } + if idSet["keep-1"] { + t.Fatalf("must not delete matched keep-1, got %q", deleteForm["file_ids"]) + } +} + // TestBatchResolve115Links 验证下载队列的批量换链：同账号多个 115 任务的 // pickcode 合并为一次 downurl 请求（官方接口支持逗号分隔多 pick_code）， // 重复引用去重、非 115 任务不参与、直链携带绑定 UA。 diff --git a/internal/service/strm_sync.go b/internal/service/strm_sync.go index 2c84904..a48b0a0 100644 --- a/internal/service/strm_sync.go +++ b/internal/service/strm_sync.go @@ -56,8 +56,10 @@ type strmSyncState struct { seenVideoTarget map[string]cloud.FileEntry activeDownloadPaths map[string]bool // 本地已在排队/进行的下载任务路径（内存去重） activeUploadPaths map[string]bool // 本地已在排队/进行的上传任务路径（内存去重） - pendingDownloads []*model.StrmDownloadTask - pendingUploads []*model.StrmUploadTask + // recentDoneUploadSizes：近期已成功上传的 local_path → size，缩短「done 但列表未到」窗口内的重复入队 + recentDoneUploadSizes map[string]int64 + pendingDownloads []*model.StrmDownloadTask + pendingUploads []*model.StrmUploadTask dirCache sync.Map // dirID (string) -> relativePath (string) dirPathToID map[string]string // relativePath (string) -> dirID（115 上传父目录寻址用，walk 后构建） dirCacheDirty map[string]string // 待批量落库的目录缓存（dirID → 相对路径），避免逐目录单条 upsert @@ -350,6 +352,14 @@ func (st *strmSyncState) run() error { } else { st.activeUploadPaths = map[string]bool{} } + since := time.Now().Add(-strmRecentUploadSkipWindow) + if recent, err := st.s.repo.StrmUpload.GetRecentDoneUploadSizeMap(st.ctx, st.p.ID, since); err == nil { + st.recentDoneUploadSizes = recent + } else { + st.recentDoneUploadSizes = map[string]int64{} + st.s.log.Warn("加载近期已完成上传任务失败，跳过 done 窗口去重", + zap.String("path_id", st.p.ID), zap.Error(err)) + } } if st.provider != nil { @@ -1575,6 +1585,7 @@ func (st *strmSyncState) walkLocalSource() error { // scanLocalMetaForUpload 扫描本地元数据，与远端比对后入上传队列。 // 以本地为准：网盘端不存在、同名不同大小、或同名同大小但 SHA1 不同（115 提供 // 远端哈希时做内容级比对）均入队覆盖上传；同名同大小同内容视为同一文件跳过。 +// 另：近期已成功上传且大小未变的路径跳过入队，缩短「任务 done 但 115 列表滞后」窗口。 func (st *strmSyncState) scanLocalMetaForUpload() error { defer st.flushPendingUploads() if st.activeUploadPaths == nil { @@ -1584,6 +1595,14 @@ func (st *strmSyncState) scanLocalMetaForUpload() error { st.activeUploadPaths = map[string]bool{} } } + if st.recentDoneUploadSizes == nil { + since := time.Now().Add(-strmRecentUploadSkipWindow) + if recent, err := st.s.repo.StrmUpload.GetRecentDoneUploadSizeMap(st.ctx, st.p.ID, since); err == nil { + st.recentDoneUploadSizes = recent + } else { + st.recentDoneUploadSizes = map[string]int64{} + } + } var pendingDeletes map[string][]string if st.p.Provider == model.StrmProvider115 { pendingDeletes = map[string][]string{} @@ -1632,23 +1651,28 @@ func (st *strmSyncState) scanLocalMetaForUpload() error { } } } - if matchedIdx >= 0 { - // 远端已存在完全一致的副本，跳过上传！ - // 若远端还存在其他同名脏副本（副本总数 > 1），在 115 下收集待删除 ID，稍后按目录批量删除。 - // 严禁将已命中的最新副本 ID (matchedID) 放入待删列表，杜绝误杀唯一有效副本。 - if len(entries) > 1 && st.p.Provider == model.StrmProvider115 { - parentCID := st.uploadRemoteTarget(rel) - if parentCID != "" { - matchedID := entries[matchedIdx].ID - for i, it := range entries { - if i != matchedIdx && it.ID != "" && it.ID != matchedID { - pendingDeletes[parentCID] = append(pendingDeletes[parentCID], it.ID) - } + if matchedIdx >= 0 { + // 远端已存在完全一致的副本，跳过上传！ + // 若远端还存在其他同名脏副本（副本总数 > 1），在 115 下收集待删除 ID，稍后按目录批量删除。 + // 严禁将已命中的最新副本 ID (matchedID) 放入待删列表，杜绝误杀唯一有效副本。 + if len(entries) > 1 && st.p.Provider == model.StrmProvider115 { + parentCID := st.uploadRemoteTarget(rel) + if parentCID != "" { + matchedID := entries[matchedIdx].ID + for i, it := range entries { + if i != matchedIdx && it.ID != "" && it.ID != matchedID { + pendingDeletes[parentCID] = append(pendingDeletes[parentCID], it.ID) } } } - return nil } + return nil + } + } + + // 近期已成功上传且大小未变：115 列表可能尚未反映，避免重复入队 + if doneSize, ok := st.recentDoneUploadSizes[path]; ok && doneSize == info.Size() { + return nil } remoteTarget := st.uploadRemoteTarget(rel) diff --git a/internal/service/strm_sync_test.go b/internal/service/strm_sync_test.go index 23dedf0..3861c1b 100644 --- a/internal/service/strm_sync_test.go +++ b/internal/service/strm_sync_test.go @@ -1453,3 +1453,120 @@ func TestWalk115AdaptiveHierarchicalFlatScan(t *testing.T) { } } } + +// TestScanLocalMetaForUploadSkipsRecentDoneSameSize 验证近期已成功上传且大小未变时， +// 即使远端列表尚未反映，也不再入队，缩短「done 但列表滞后」窗口。 +func TestScanLocalMetaForUploadSkipsRecentDoneSameSize(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + nfoPath := filepath.Join(localDir, "movie.nfo") + content := []byte("uploaded-recently") + writeFile(t, nfoPath, string(content)) + + p := &model.StrmSyncPath{ + Base: model.Base{ID: "recent-done-skip"}, + Provider: model.StrmProvider115, + RemotePath: "root-cid", + LocalPath: localDir, + UploadMeta: true, + } + now := time.Now() + doneTask := &model.StrmUploadTask{ + SyncPathID: p.ID, + Provider: model.StrmProvider115, + FileName: "movie.nfo", + LocalPath: nfoPath, + RemotePath: "parent-cid", + RemoteRef: "new-file-id", + Size: int64(len(content)), + Status: model.StrmTaskDone, + FinishedAt: &now, + } + if err := svc.repo.StrmUpload.Create(context.Background(), doneTask); err != nil { + t.Fatal(err) + } + + st := &strmSyncState{ + s: svc, + ctx: context.Background(), + p: p, + cfg: &strmPathConfig{UploadMeta: true, MetaExt: []string{"nfo"}}, + rec: &model.StrmSyncRecord{}, + seenMeta: map[string]bool{}, + remoteMeta: map[string][]remoteMetaItem{}, // 远端列表空 = 滞后 + dirPathToID: map[string]string{"": "parent-cid"}, + activeUploadPaths: map[string]bool{}, + recentDoneUploadSizes: nil, // 让 scan 自行从 DB 加载 + } + if err := st.scanLocalMetaForUpload(); err != nil { + t.Fatalf("scanLocalMetaForUpload failed: %v", err) + } + if st.rec.Uploaded != 0 { + t.Fatalf("recent done same-size should skip enqueue, uploaded=%d", st.rec.Uploaded) + } + if len(st.pendingUploads) != 0 { + t.Fatalf("expected no pending uploads, got %d", len(st.pendingUploads)) + } +} + +// TestScanLocalMetaForUploadRequeuesWhenRecentDoneSizeChanged 本地大小变化后不应被近期 done 窗口挡住。 +func TestScanLocalMetaForUploadRequeuesWhenRecentDoneSizeChanged(t *testing.T) { + svc := testStrmService(t) + localDir := t.TempDir() + nfoPath := filepath.Join(localDir, "movie.nfo") + writeFile(t, nfoPath, "new-longer-content-xxx") + + p := &model.StrmSyncPath{ + Base: model.Base{ID: "recent-done-resize"}, + Provider: model.StrmProvider115, + RemotePath: "root-cid", + LocalPath: localDir, + UploadMeta: true, + } + now := time.Now() + doneTask := &model.StrmUploadTask{ + SyncPathID: p.ID, + Provider: model.StrmProvider115, + FileName: "movie.nfo", + LocalPath: nfoPath, + RemotePath: "parent-cid", + Size: 3, // 旧大小 + Status: model.StrmTaskDone, + FinishedAt: &now, + } + if err := svc.repo.StrmUpload.Create(context.Background(), doneTask); err != nil { + t.Fatal(err) + } + + st := &strmSyncState{ + s: svc, + ctx: context.Background(), + p: p, + cfg: &strmPathConfig{UploadMeta: true, MetaExt: []string{"nfo"}}, + rec: &model.StrmSyncRecord{}, + seenMeta: map[string]bool{}, + remoteMeta: map[string][]remoteMetaItem{}, + dirPathToID: map[string]string{"": "parent-cid"}, + activeUploadPaths: map[string]bool{}, + } + if err := st.scanLocalMetaForUpload(); err != nil { + t.Fatalf("scanLocalMetaForUpload failed: %v", err) + } + if st.rec.Uploaded != 1 { + t.Fatalf("size changed should re-enqueue, uploaded=%d", st.rec.Uploaded) + } + tasks, _, err := svc.repo.StrmUpload.List(context.Background(), model.StrmTaskPending, 1, 20) + if err != nil { + t.Fatal(err) + } + found := false + for _, task := range tasks { + if task.LocalPath == nfoPath && task.Status == model.StrmTaskPending { + found = true + break + } + } + if !found { + t.Fatal("expected a pending upload task for resized local file") + } +}