mirror of
https://github.com/truewhile/MeBox.git
synced 2026-10-06 21:36:37 +08:00
Fix license sync and cloud scan scheduling
This commit is contained in:
+83
-10
@@ -682,13 +682,19 @@ func (s *ScannerService) CancelCloudScan(libraryID string) bool {
|
||||
s.cloudScanMu.Lock()
|
||||
defer s.cloudScanMu.Unlock()
|
||||
entry := s.cloudScans[libraryID]
|
||||
if entry == nil || entry.cancel == nil || (entry.status.State != "running" && entry.status.State != "canceling") {
|
||||
if entry == nil || (entry.status.State != "running" && entry.status.State != "queued" && entry.status.State != "canceling") {
|
||||
return false
|
||||
}
|
||||
entry.status.State = "canceling"
|
||||
entry.status.Stage = "canceling"
|
||||
entry.status.UpdatedAt = time.Now()
|
||||
entry.cancel()
|
||||
if entry.cancel != nil {
|
||||
entry.cancel()
|
||||
} else {
|
||||
entry.status.State = "canceled"
|
||||
entry.status.Stage = "canceled"
|
||||
entry.status.FinishedAt = time.Now()
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -700,13 +706,19 @@ func (s *ScannerService) CancelAllCloudScans() int {
|
||||
defer s.cloudScanMu.Unlock()
|
||||
cancelled := 0
|
||||
for _, entry := range s.cloudScans {
|
||||
if entry == nil || entry.cancel == nil || (entry.status.State != "running" && entry.status.State != "canceling") {
|
||||
if entry == nil || (entry.status.State != "running" && entry.status.State != "queued" && entry.status.State != "canceling") {
|
||||
continue
|
||||
}
|
||||
entry.status.State = "canceling"
|
||||
entry.status.Stage = "canceling"
|
||||
entry.status.UpdatedAt = time.Now()
|
||||
entry.cancel()
|
||||
if entry.cancel != nil {
|
||||
entry.cancel()
|
||||
} else {
|
||||
entry.status.State = "canceled"
|
||||
entry.status.Stage = "canceled"
|
||||
entry.status.FinishedAt = time.Now()
|
||||
}
|
||||
cancelled++
|
||||
}
|
||||
return cancelled
|
||||
@@ -724,13 +736,19 @@ func (s *ScannerService) CancelCloudScansForProvider(provider string) int {
|
||||
defer s.cloudScanMu.Unlock()
|
||||
cancelled := 0
|
||||
for _, entry := range s.cloudScans {
|
||||
if entry == nil || entry.status.Provider != provider || entry.cancel == nil || (entry.status.State != "running" && entry.status.State != "canceling") {
|
||||
if entry == nil || entry.status.Provider != provider || (entry.status.State != "running" && entry.status.State != "queued" && entry.status.State != "canceling") {
|
||||
continue
|
||||
}
|
||||
entry.status.State = "canceling"
|
||||
entry.status.Stage = "canceling"
|
||||
entry.status.UpdatedAt = time.Now()
|
||||
entry.cancel()
|
||||
if entry.cancel != nil {
|
||||
entry.cancel()
|
||||
} else {
|
||||
entry.status.State = "canceled"
|
||||
entry.status.Stage = "canceled"
|
||||
entry.status.FinishedAt = time.Now()
|
||||
}
|
||||
cancelled++
|
||||
}
|
||||
return cancelled
|
||||
@@ -816,22 +834,77 @@ func (s *ScannerService) StartAllCloudLibraryScans() ([]CloudScanStatus, error)
|
||||
}
|
||||
libs = FilterScannableCloudLibraries(context.Background(), s.repo, libs)
|
||||
statuses := make([]CloudScanStatus, 0, len(libs))
|
||||
queue := make([]string, 0, len(libs))
|
||||
for _, lib := range libs {
|
||||
if !lib.Enabled {
|
||||
continue
|
||||
}
|
||||
if _, ok := ParseCloudLibraryMount(lib.Path); !ok {
|
||||
mount, ok := ParseCloudLibraryMount(lib.Path)
|
||||
if !ok {
|
||||
continue
|
||||
}
|
||||
status, _, err := s.StartCloudLibraryScan(lib.ID, false)
|
||||
if err != nil {
|
||||
status = CloudScanStatus{LibraryID: lib.ID, State: "error", Error: err.Error(), UpdatedAt: time.Now()}
|
||||
status, queued := s.queueCloudLibraryScan(lib, mount)
|
||||
if queued {
|
||||
queue = append(queue, lib.ID)
|
||||
}
|
||||
statuses = append(statuses, status)
|
||||
}
|
||||
if len(queue) > 0 {
|
||||
go s.runQueuedCloudLibraryScans(queue)
|
||||
}
|
||||
return statuses, nil
|
||||
}
|
||||
|
||||
func (s *ScannerService) queueCloudLibraryScan(lib model.Library, mount CloudMountInfo) (CloudScanStatus, bool) {
|
||||
now := time.Now()
|
||||
status := CloudScanStatus{
|
||||
LibraryID: lib.ID,
|
||||
Provider: mount.Provider,
|
||||
Stage: "queued",
|
||||
State: "queued",
|
||||
StartedAt: now,
|
||||
UpdatedAt: now,
|
||||
ResumeHint: "中断后再次点击扫描会从头遍历,但已入库媒体会去重更新,只补齐缺失项。",
|
||||
Estimate: "小目录通常几十秒;几万文件的大目录可能需要数分钟到数小时,取决于网盘接口速度。",
|
||||
}
|
||||
s.cloudScanMu.Lock()
|
||||
defer s.cloudScanMu.Unlock()
|
||||
if s.cloudScans == nil {
|
||||
s.cloudScans = make(map[string]*cloudScanEntry)
|
||||
}
|
||||
if entry := s.cloudScans[lib.ID]; entry != nil {
|
||||
switch entry.status.State {
|
||||
case "running", "queued", "canceling":
|
||||
return entry.status, false
|
||||
}
|
||||
}
|
||||
s.cloudScans[lib.ID] = &cloudScanEntry{status: status}
|
||||
return status, true
|
||||
}
|
||||
|
||||
func (s *ScannerService) runQueuedCloudLibraryScans(libraryIDs []string) {
|
||||
ctx, cancel := cloudScanContext(context.Background(), cloudScanTimeout(context.Background(), s.repo, 24*time.Hour))
|
||||
defer cancel()
|
||||
for _, libraryID := range libraryIDs {
|
||||
if ctx.Err() != nil {
|
||||
return
|
||||
}
|
||||
if s.cloudScanWasCanceled(libraryID) {
|
||||
continue
|
||||
}
|
||||
if _, err := s.ScanLibraryWithoutAutoScrape(ctx, libraryID); err != nil && !errors.Is(err, ErrCloudScanAlreadyRunning) && !errors.Is(err, context.Canceled) && s.log != nil {
|
||||
s.log.Warn("cloud library queued scan failed", zap.String("library_id", libraryID), zap.Error(err))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (s *ScannerService) cloudScanWasCanceled(libraryID string) bool {
|
||||
s.cloudScanMu.Lock()
|
||||
defer s.cloudScanMu.Unlock()
|
||||
entry := s.cloudScans[libraryID]
|
||||
return entry != nil && entry.status.State == "canceled"
|
||||
}
|
||||
|
||||
// ScanLibrary walks the library root and persists discovered media files.
|
||||
func (s *ScannerService) ScanLibrary(ctx context.Context, libraryID string) (*ScanResult, error) {
|
||||
return s.scanLibrary(ctx, libraryID, true)
|
||||
|
||||
@@ -87,8 +87,8 @@ const (
|
||||
cloudAutoSyncEnabledKey = "cloud.auto_sync_enabled"
|
||||
cloudSyncIntervalSecondsKey = "cloud.sync_interval_seconds"
|
||||
cloudLastAutoSyncDateKey = "cloud.last_auto_sync_date"
|
||||
cloudAutoSyncWindowStartHour = 19
|
||||
cloudAutoSyncWindowEndHour = 21
|
||||
cloudAutoSyncWindowStartHour = 23
|
||||
cloudAutoSyncWindowEndHour = 5
|
||||
cloudAutoSyncCompletedDateForm = "2006-01-02"
|
||||
)
|
||||
|
||||
@@ -498,19 +498,33 @@ func (s *SchedulerService) autoCloudSyncDue(ctx context.Context, now time.Time)
|
||||
if err != nil {
|
||||
return true
|
||||
}
|
||||
return strings.TrimSpace(last) != now.Format(cloudAutoSyncCompletedDateForm)
|
||||
return strings.TrimSpace(last) != cloudAutoSyncWindowDate(now)
|
||||
}
|
||||
|
||||
func cloudAutoSyncInWindow(now time.Time) bool {
|
||||
hour := now.In(time.Local).Hour()
|
||||
return hour >= cloudAutoSyncWindowStartHour && hour < cloudAutoSyncWindowEndHour
|
||||
if cloudAutoSyncWindowStartHour == cloudAutoSyncWindowEndHour {
|
||||
return true
|
||||
}
|
||||
if cloudAutoSyncWindowStartHour < cloudAutoSyncWindowEndHour {
|
||||
return hour >= cloudAutoSyncWindowStartHour && hour < cloudAutoSyncWindowEndHour
|
||||
}
|
||||
return hour >= cloudAutoSyncWindowStartHour || hour < cloudAutoSyncWindowEndHour
|
||||
}
|
||||
|
||||
func cloudAutoSyncWindowDate(now time.Time) string {
|
||||
local := now.In(time.Local)
|
||||
if cloudAutoSyncWindowStartHour > cloudAutoSyncWindowEndHour && local.Hour() < cloudAutoSyncWindowEndHour {
|
||||
local = local.AddDate(0, 0, -1)
|
||||
}
|
||||
return local.Format(cloudAutoSyncCompletedDateForm)
|
||||
}
|
||||
|
||||
func (s *SchedulerService) markCloudAutoSyncCompleted(ctx context.Context, now time.Time) error {
|
||||
if s.repo == nil || s.repo.Setting == nil {
|
||||
return nil
|
||||
}
|
||||
return s.repo.Setting.Set(ctx, cloudLastAutoSyncDateKey, now.Format(cloudAutoSyncCompletedDateForm))
|
||||
return s.repo.Setting.Set(ctx, cloudLastAutoSyncDateKey, cloudAutoSyncWindowDate(now))
|
||||
}
|
||||
|
||||
func (s *SchedulerService) cloudSyncInterval(ctx context.Context) time.Duration {
|
||||
|
||||
@@ -332,7 +332,7 @@ func TestSchedulerCloudSyncRunsOnlyOnceInsideNightlyWindow(t *testing.T) {
|
||||
scheduler := NewSchedulerService(log, repos, scanner, nil, nil, storage, NewHub(log), "")
|
||||
|
||||
scheduler.now = func() time.Time {
|
||||
return time.Date(2026, 6, 11, 18, 30, 0, 0, time.Local)
|
||||
return time.Date(2026, 6, 11, 22, 30, 0, 0, time.Local)
|
||||
}
|
||||
if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil {
|
||||
t.Fatalf("cloud sync outside window: %v", err)
|
||||
@@ -353,13 +353,26 @@ func TestSchedulerCloudSyncRunsOnlyOnceInsideNightlyWindow(t *testing.T) {
|
||||
}
|
||||
|
||||
scheduler.now = func() time.Time {
|
||||
return time.Date(2026, 6, 11, 20, 15, 0, 0, time.Local)
|
||||
return time.Date(2026, 6, 12, 4, 15, 0, 0, time.Local)
|
||||
}
|
||||
if !cloudAutoSyncInWindow(scheduler.now()) {
|
||||
t.Fatalf("04:15 should still be inside overnight cloud sync window")
|
||||
}
|
||||
if got := cloudAutoSyncWindowDate(scheduler.now()); got != "2026-06-11" {
|
||||
t.Fatalf("04:15 should belong to previous nightly window, got %s", got)
|
||||
}
|
||||
if err := scheduler.jobSyncCloudLibraries(t.Context()); err != nil {
|
||||
t.Fatalf("second cloud sync same day: %v", err)
|
||||
t.Fatalf("second cloud sync same overnight window: %v", err)
|
||||
}
|
||||
if got := requests.Load(); got != 1 {
|
||||
t.Fatalf("same-day auto sync should not rerun, requests = %d", got)
|
||||
t.Fatalf("same overnight auto sync should not rerun, requests = %d", got)
|
||||
}
|
||||
|
||||
scheduler.now = func() time.Time {
|
||||
return time.Date(2026, 6, 12, 5, 0, 0, 0, time.Local)
|
||||
}
|
||||
if cloudAutoSyncInWindow(scheduler.now()) {
|
||||
t.Fatalf("05:00 should be outside overnight cloud sync window")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -465,7 +478,7 @@ func TestSchedulerCloudSyncDisabledByDefault(t *testing.T) {
|
||||
}
|
||||
|
||||
func fixedNightlySyncTime() time.Time {
|
||||
return time.Date(2026, 6, 11, 19, 30, 0, 0, time.Local)
|
||||
return time.Date(2026, 6, 11, 23, 30, 0, 0, time.Local)
|
||||
}
|
||||
|
||||
func TestSchedulerLoopWaitsIntervalAfterSlowRun(t *testing.T) {
|
||||
|
||||
Reference in New Issue
Block a user