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

优化同步逻辑

Browse Source 
优化同步逻辑
This commit is contained in:
truewhile
2026-09-06 16:17:42 +08:00
committed by GitHub
parent a71a18ce82 3150971f10
commit b4a1dc38bb
2 changed files with 421 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/service/strm_sync.go
+311 -102
View File
@@ -514,28 +514,31 @@ func (st *strmSyncState) walkRemote() error {
if task.rel != "" {
rel = task.rel + "/" + cleanName
}
if entry.IsDir {
push(dirTask{id: entry.ID, rel: rel})
} else {
st.processRemoteFile(entry, rel)
if entry.IsDir {
st.dirCache.Store(entry.ID, rel)
st.deferDirCacheSave(entry.ID, rel)
push(dirTask{id: entry.ID, rel: rel})
} else {
st.processRemoteFile(entry, rel)
}
}
walkMu.Lock()
pending--
if pending == 0 {
walkCond.Broadcast()
}
walkMu.Unlock()
}
walkMu.Lock()
pending--
if pending == 0 {
walkCond.Broadcast()
}
walkMu.Unlock()
}); err != nil {
cancel()
}
}); err != nil {
cancel()
}
}()
}
wg.Wait()
if firstErr != nil {
return firstErr
}
}()
}
wg.Wait()
st.flushDirCacheSave()
if firstErr != nil {
return firstErr
}
return ctx.Err()
}
@@ -674,93 +677,16 @@ func (st *strmSyncState) walk115Flat(open115 *cloud115.OpenClient) error {
if pathCounts[item.Path] > 1 {
continue
}
st.dirCache.Store(item.DirID, cleanDirRel(item.Path))
st.dirCache.Store(item.DirID, cleanDirRel(item.Path))
}
}
}
}
// 2. 探测文件总数
const pageSize = 1150
firstBatch, totalCount, err := open115.GetFsListFlat(ctx, rootCID, 0, pageSize)
if err != nil {
return fmt.Errorf("115: 获取文件列表失败：%w", err)
}
// 115 的扁平化列表（搜索底层）对 offset + limit 有 10000 的最大深度限制。
// 当扁平模式下的总文件数 >= 9500 时，强行拒绝继续扁平拉取，而是抛出降级错误，
// 让外层回退到使用普通的按目录并发递归（walkRemote），以免截断导致后排文件被误删/重传。
if totalCount >= 9500 {
return errFallbackToWalkRemote
}
st.updateSyncMessage(fmt.Sprintf("正在拉取远端文件列表 (共 %d 个文件)...", totalCount))
allFiles := make([]cloud115.RemoteFile, 0, totalCount)
allFiles = append(allFiles, firstBatch...)
// 3. 并发分页拉取剩余文件
if totalCount > int64(len(firstBatch)) {
totalPages := int((totalCount + pageSize - 1) / pageSize)
type pageTask struct {
offset int
// 2. 自适应分治拉取文件列表（单目录超 9500 时自动对子目录并发分治扁平化）
allFiles, err := st.fetch115FilesAdaptive(ctx, open115, rootCID)
if err != nil {
return err
}
pageTasks := make([]pageTask, 0, totalPages-1)
for page := 1; page < totalPages; page++ {
pageTasks = append(pageTasks, pageTask{offset: page * pageSize})
}
var (
filesMu sync.Mutex
wg sync.WaitGroup
taskCh = make(chan pageTask, len(pageTasks))
errMu sync.Mutex
fetchErr error
)
for _, t := range pageTasks {
taskCh <- t
}
close(taskCh)
workers := 8
if len(pageTasks) < workers {
workers = len(pageTasks)
}
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
// 分页拉取 panic 时取消整个同步；正常退出不取消。
if err := helper.Recover(st.s.log, "strm.sync.walk115.page", func() error {
for t := range taskCh {
if ctx.Err() != nil {
return nil
}
files, _, err := open115.GetFsListFlat(ctx, rootCID, t.offset, pageSize)
if err != nil {
errMu.Lock()
if fetchErr == nil {
fetchErr = err
}
errMu.Unlock()
return nil
}
filesMu.Lock()
allFiles = append(allFiles, files...)
filesMu.Unlock()
}
return nil
}); err != nil {
cancel()
}
}()
}
wg.Wait()
if fetchErr != nil {
return fmt.Errorf("115: 分页拉取失败：%w", fetchErr)
}
}
if ctx.Err() != nil {
return ctx.Err()
@@ -976,6 +902,289 @@ feed:
return ctx.Err()
}
const (
flat115PageSize = 1150
flat115Threshold = 9500
max115AdaptiveDepth = 10
)
type adaptive115Task struct {
cid string
rel string
depth int
}
// fetch115FlatSubtree 扁平拉取单个文件数在安全阈值内的子树全部文件。
func fetch115FlatSubtree(ctx context.Context, open115 *cloud115.OpenClient, cid string, firstBatch []cloud115.RemoteFile, totalCount int64, log *zap.Logger) ([]cloud115.RemoteFile, error) {
allFiles := make([]cloud115.RemoteFile, 0, totalCount)
allFiles = append(allFiles, firstBatch...)
if totalCount <= int64(len(firstBatch)) {
return allFiles, nil
}
totalPages := int((totalCount + flat115PageSize - 1) / flat115PageSize)
type pageTask struct {
offset int
}
pageTasks := make([]pageTask, 0, totalPages-1)
for page := 1; page < totalPages; page++ {
pageTasks = append(pageTasks, pageTask{offset: page * flat115PageSize})
}
var (
filesMu sync.Mutex
wg sync.WaitGroup
taskCh = make(chan pageTask, len(pageTasks))
errMu sync.Mutex
fetchErr error
)
for _, t := range pageTasks {
taskCh <- t
}
close(taskCh)
workers := 8
if len(pageTasks) < workers {
workers = len(pageTasks)
}
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if err := helper.Recover(log, "strm.sync.walk115.page", func() error {
for t := range taskCh {
if ctx.Err() != nil {
return nil
}
files, _, err := open115.GetFsListFlat(ctx, cid, t.offset, flat115PageSize)
if err != nil {
errMu.Lock()
if fetchErr == nil {
fetchErr = err
}
errMu.Unlock()
return nil
}
filesMu.Lock()
allFiles = append(allFiles, files...)
filesMu.Unlock()
}
return nil
}); err != nil {
errMu.Lock()
if fetchErr == nil {
fetchErr = err
}
errMu.Unlock()
}
}()
}
wg.Wait()
if fetchErr != nil {
return nil, fmt.Errorf("115: 分页拉取失败：%w", fetchErr)
}
return allFiles, nil
}
// list115DirDirect 列出指定目录下的直接子项（单层 cur=1&show_dir=1）。
func list115DirDirect(ctx context.Context, open115 *cloud115.OpenClient, cid string) ([]cloud115.RemoteFile, error) {
var out []cloud115.RemoteFile
for offset := 0; ; offset += flat115PageSize {
files, _, err := open115.GetFsList(ctx, cid, offset, flat115PageSize)
if err != nil {
return nil, err
}
out = append(out, files...)
if len(files) < flat115PageSize {
break
}
select {
case <-ctx.Done():
return nil, ctx.Err()
default:
}
}
return out, nil
}
// fetch115FilesAdaptive 采用自适应分治策略抓取 115 目录树下的全部文件：
// 115 开放平台扁平搜索对 offset+limit 有 10000 的最大深度限制。
// - 若子树文件总数 < 9500，直接使用全速扁平分页批量拉取；
// - 若子树文件总数 >= 9500（大库或超大分类目录），自动分治：仅单层列出该目录的直属子项（cur=1），
// 直属纯文件直接收集，直属子目录则派发为独立的子树任务继续递归探测与拉取；
// - 若超大单目录下无子目录或层级过深（>10层），安全回退到 errFallbackToWalkRemote。
func (st *strmSyncState) fetch115FilesAdaptive(ctx context.Context, open115 *cloud115.OpenClient, rootCID string) ([]cloud115.RemoteFile, error) {
var (
allFiles []cloud115.RemoteFile
filesMu sync.Mutex
walkMu sync.Mutex
walkCond = sync.NewCond(&walkMu)
work []adaptive115Task
pending int
errMu sync.Mutex
firstErr error
)
push := func(t adaptive115Task) {
walkMu.Lock()
work = append(work, t)
pending++
walkCond.Signal()
walkMu.Unlock()
}
go func() {
<-ctx.Done()
walkMu.Lock()
walkCond.Broadcast()
walkMu.Unlock()
}()
push(adaptive115Task{cid: rootCID, rel: "", depth: 0})
workers := 8
var wg sync.WaitGroup
for i := 0; i < workers; i++ {
wg.Add(1)
go func() {
defer wg.Done()
if err := helper.Recover(st.s.log, "strm.sync.walk115.adaptive", func() error {
for {
walkMu.Lock()
for len(work) == 0 {
if ctx.Err() != nil || pending == 0 {
walkMu.Unlock()
return nil
}
walkCond.Wait()
}
task := work[0]
work = work[1:]
walkMu.Unlock()
if ctx.Err() != nil {
walkMu.Lock()
pending--
walkCond.Broadcast()
walkMu.Unlock()
return nil
}
firstBatch, totalCount, err := open115.GetFsListFlat(ctx, task.cid, 0, flat115PageSize)
if err != nil {
errMu.Lock()
if firstErr == nil {
firstErr = fmt.Errorf("115: 获取文件列表失败（cid=%s）：%w", task.cid, err)
}
errMu.Unlock()
walkMu.Lock()
pending--
walkCond.Broadcast()
walkMu.Unlock()
return nil
}
if totalCount < flat115Threshold {
// 安全深度内：直接扁平拉取该子树全部文件
files, err := fetch115FlatSubtree(ctx, open115, task.cid, firstBatch, totalCount, st.s.log)
if err != nil {
errMu.Lock()
if firstErr == nil {
firstErr = err
}
errMu.Unlock()
walkMu.Lock()
pending--
walkCond.Broadcast()
walkMu.Unlock()
return nil
}
filesMu.Lock()
allFiles = append(allFiles, files...)
currentCount := len(allFiles)
filesMu.Unlock()
st.updateSyncMessage(fmt.Sprintf("正在拉取远端文件列表 (已获取 %d 个文件)...", currentCount))
} else {
// 子树过大（>=9500）：分治展开该目录直接子项
if task.depth >= max115AdaptiveDepth {
// 深度超限兜底：单目录嵌套超 10 层仍超 9500，回退为传统递归
errMu.Lock()
if firstErr == nil {
firstErr = errFallbackToWalkRemote
}
errMu.Unlock()
walkMu.Lock()
pending--
walkCond.Broadcast()
walkMu.Unlock()
return nil
}
st.s.log.Info("115: 目录文件数超限，自动分治展开子目录并发扁平拉取",
zap.String("cid", task.cid),
zap.String("rel", task.rel),
zap.Int64("total_count", totalCount),
zap.Int("depth", task.depth))
directEntries, err := list115DirDirect(ctx, open115, task.cid)
if err != nil {
errMu.Lock()
if firstErr == nil {
firstErr = fmt.Errorf("115: 列出单层目录失败（cid=%s）：%w", task.cid, err)
}
errMu.Unlock()
walkMu.Lock()
pending--
walkCond.Broadcast()
walkMu.Unlock()
return nil
}
for _, f := range directEntries {
if f.Category == cloud115.TypeDir {
cleanName := cleanEntryName(f.FileName, true)
subRel := cleanName
if task.rel != "" {
subRel = task.rel + "/" + cleanName
}
st.dirCache.Store(f.FileId, subRel)
st.deferDirCacheSave(f.FileId, subRel)
push(adaptive115Task{cid: f.FileId, rel: subRel, depth: task.depth + 1})
} else {
filesMu.Lock()
allFiles = append(allFiles, f)
filesMu.Unlock()
}
}
}
walkMu.Lock()
pending--
if pending == 0 {
walkCond.Broadcast()
}
walkMu.Unlock()
}
}); err != nil {
errMu.Lock()
if firstErr == nil {
firstErr = err
}
errMu.Unlock()
}
}()
}
wg.Wait()
if firstErr != nil {
return nil, firstErr
}
return allFiles, nil
}
// handleVideo 生成/更新 .strm 文件。
func (st *strmSyncState) handleVideo(entry cloud.FileEntry, rel, ext string) {
relSansExt := rel[:len(rel)-len(ext)]
internal/service/strm_sync_test.go
+110
View File
@@ -1319,3 +1319,113 @@ func TestScanLocalMetaForUploadMultipleCopiesAllStale(t *testing.T) {
t.Fatalf("expected RemoteRef to be 'old-1,old-2', got %q", tasks[0].RemoteRef)
}
}
// TestWalk115AdaptiveHierarchicalFlatScan 验证自适应分治扁平化扫描：
// 当根目录探测总数 >= 9500 时，系统自动分治展开单层直接子项，对各子目录分别执行扁平拉取，
// 正确合并根目录直属文件与各子目录深层文件，突破 115 开放平台 10000 深度限制。
func TestWalk115AdaptiveHierarchicalFlatScan(t *testing.T) {
svc := testStrmService(t)
localDir := t.TempDir()
acct := &model.StrmAccount{Name: "fake115-adaptive", Provider: "cloud115", Config: "{}", Enabled: true}
if err := svc.repo.StrmAccount.Create(context.Background(), acct); err != nil {
t.Fatal(err)
}
p := &model.StrmSyncPath{
Base: model.Base{ID: "adaptive-path"},
AccountID: acct.ID,
Provider: model.StrmProvider115,
RemotePath: "0",
LocalPath: localDir,
}
api := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
q := r.URL.Query()
cid := q.Get("cid")
cur := q.Get("cur")
switch r.URL.Path {
case "/open/ufile/files":
if cid == "0" && cur == "0" {
// 根目录扁平探测：模拟总文件数 12000 超限 (>= 9500)
w.Write([]byte(`{"state":true,"count":12000,"data":[]}`))
return
}
if cid == "0" && cur == "1" {
// 根目录单层列举：返回 1 个直属视频和 2 个子目录
w.Write([]byte(`{"state":true,"count":3,"data":[
{"fid":"100","pid":"0","fc":"1","fn":"root.mkv","pc":"pcr","upt":1700000000,"fs":1024},
{"fid":"1001","pid":"0","fc":"0","fn":"Heyzo","upt":1700000001,"fs":0},
{"fid":"1002","pid":"0","fc":"0","fn":"S1","upt":1700000002,"fs":0}]}`))
return
}
if cid == "1001" {
// 子目录 Heyzo 扁平拉取：文件数安全 (< 9500)
w.Write([]byte(`{"state":true,"count":2,"data":[
{"fid":"201","pid":"1001","fc":"1","fn":"h1.mkv","pc":"pc201","upt":1700000001,"fs":1024},
{"fid":"202","pid":"1001","fc":"1","fn":"h2.mkv","pc":"pc202","upt":1700000002,"fs":2048}]}`))
return
}
if cid == "1002" {
// 子目录 S1 扁平拉取：文件数安全 (< 9500)
w.Write([]byte(`{"state":true,"count":1,"data":[
{"fid":"301","pid":"1002","fc":"1","fn":"s1.mkv","pc":"pc301","upt":1700000003,"fs":4096}]}`))
return
}
t.Errorf("unexpected files query: %s", r.URL.RawQuery)
case "/open/folder/get_info":
fileID := q.Get("file_id")
switch fileID {
case "1001":
w.Write([]byte(`{"state":true,"data":{"file_id":"1001","file_name":"Heyzo","file_category":"0",
"paths":[{"file_id":"0","file_name":"根目录"},{"file_id":"1001","file_name":"Heyzo"}]}}`))
case "1002":
w.Write([]byte(`{"state":true,"data":{"file_id":"1002","file_name":"S1","file_category":"0",
"paths":[{"file_id":"0","file_name":"根目录"},{"file_id":"1002","file_name":"S1"}]}}`))
default:
t.Errorf("unexpected get_info file_id: %s", fileID)
}
default:
t.Errorf("unexpected path %s", r.URL.Path)
}
}))
defer api.Close()
oldPro := cloud115.ProAPIBase
cloud115.ProAPIBase = api.URL
defer func() { cloud115.ProAPIBase = oldPro }()
oc := cloud115.NewOpenClient("app", "at", "rt")
st := &strmSyncState{
s: svc,
ctx: context.Background(),
p: p,
provider: cloud.NewOpenAPI115("app", "at", "rt"),
cfg: &strmPathConfig{VideoExt: []string{"mkv"}, MetaExt: []string{"nfo"}, AddPath: 1},
rec: &model.StrmSyncRecord{},
syncType: model.StrmSyncTypeFull,
dirCache: sync.Map{},
seenVideo: map[string]bool{},
seenMeta: map[string]bool{},
remoteMeta: map[string][]remoteMetaItem{},
seenMetaTarget: map[string]cloud.FileEntry{},
seenVideoTarget: map[string]cloud.FileEntry{},
}
if err := st.walk115Flat(oc); err != nil {
t.Fatalf("walk115Flat adaptive failed: %v", err)
}
// 1个根目录视频 + 2个Heyzo视频 + 1个S1视频 = 共4个视频成功生成 .strm
if st.rec.NewStrm != 4 {
t.Fatalf("expected 4 strm created, got %d", st.rec.NewStrm)
}
expectedFiles := []string{
filepath.Join(localDir, "root.strm"),
filepath.Join(localDir, "Heyzo", "h1.strm"),
filepath.Join(localDir, "Heyzo", "h2.strm"),
filepath.Join(localDir, "S1", "s1.strm"),
}
for _, f := range expectedFiles {
if _, err := os.Stat(f); err != nil {
t.Fatalf("expected strm %s to exist, err: %v", f, err)
}
}
}