tcjlj/MeBox
1
0
Fork 0
mirror of https://github.com/truewhile/MeBox.git synced 2026-09-28 03:06:38 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(strm): 115 元数据上传幂等，避免重复上传 (#29)

Browse Source 
上传/重试前按父目录+文件名+SHA1 探活，已有同内容副本则跳过并清脏副本；
成功后回写新 file_id；同步扫描认可 30 分钟内同路径同大小的 done 任务，
缩短「任务已完成但列表未反映」导致的再入队窗口。

Co-authored-by: Cursor Agent <cursoragent@cursor.com>
Co-authored-by: truewhile <truewhile@users.noreply.github.com>
This commit is contained in:
truewhile
2026-09-07 09:01:58 +08:00
committed by GitHub
parent 2e636ce43f
commit 6bf8a6878a
8 changed files with 501 additions and 72 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/model/strm.go
+1 -1
View File
@@ -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"`
internal/repository/strm_repository.go
+28
View File
@@ -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 < ?",
internal/service/cloud115/open.go
+54
View File
@@ -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}
internal/service/cloud115/open_find_test.go
+54
View File
@@ -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))
}
}
internal/service/strm_queue.go
+104 -52
View File
@@ -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) {
internal/service/strm_queue_test.go
+104 -4
View File
@@ -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。
internal/service/strm_sync.go
+39 -15
View File
@@ -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)
internal/service/strm_sync_test.go
+117
View File
@@ -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")
}
}