fix: 防止启动整理扫描阻塞登录

This commit is contained in:
ShukeBta
2026-06-12 19:06:47 +08:00
parent 1568ae127a
commit d270e9034f
11 changed files with 312 additions and 34 deletions
+1 -1
View File
@@ -45,7 +45,7 @@ func loginHandler(svc *service.Container) gin.HandlerFunc {
"user": resp.User,
"tokens": resp.Tokens,
})
svc.Audit.Record(c.Request.Context(), resp.User.ID, "auth.login", resp.User.Username, c.ClientIP(), "")
svc.Audit.RecordBestEffort(resp.User.ID, "auth.login", resp.User.Username, c.ClientIP(), "")
}
}
+14
View File
@@ -8,6 +8,7 @@ package service
import (
"context"
"time"
"go.uber.org/zap"
@@ -39,3 +40,16 @@ func (a *AuditService) Record(ctx context.Context, userID, action, target, ip, d
a.log.Debug("audit write failed", zap.Error(err))
}
}
// RecordBestEffort writes an audit row off the request path. Login must not be
// held open by SQLite write pressure from scans or background maintenance.
func (a *AuditService) RecordBestEffort(userID, action, target, ip, detail string) {
if a == nil || a.repo == nil || a.repo.Log == nil {
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
a.Record(ctx, userID, action, target, ip, detail)
}()
}
+14 -1
View File
@@ -163,10 +163,23 @@ func (s *AuthService) Login(ctx context.Context, username, password string) (*Lo
if err != nil {
return nil, err
}
_ = s.repo.User.TouchLogin(ctx, u.ID)
s.touchLoginBestEffort(u.ID)
return &LoginResponse{User: u, Tokens: tokens}, nil
}
func (s *AuthService) touchLoginBestEffort(userID string) {
if s == nil || s.repo == nil || s.repo.User == nil || strings.TrimSpace(userID) == "" {
return
}
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
if err := s.repo.User.TouchLogin(ctx, userID); err != nil && s.log != nil {
s.log.Debug("touch login delayed", zap.String("user_id", userID), zap.Error(err))
}
}()
}
// ChangePassword updates the user password if the old one matches.
func (s *AuthService) ChangePassword(ctx context.Context, userID, oldPwd, newPwd string) error {
if strings.TrimSpace(newPwd) == "" || len(newPwd) < 6 {
@@ -30,6 +30,11 @@ func newAuthTestServices(t *testing.T) (*repository.Container, *AuthService, *Pr
if err := db.AutoMigrate(&model.User{}, &model.UserPermission{}, &model.RefreshToken{}, &model.TelegramBinding{}, &model.Setting{}); err != nil {
t.Fatal(err)
}
sqlDB, err := db.DB()
if err != nil {
t.Fatal(err)
}
sqlDB.SetMaxOpenConns(1)
repos := repository.New(db)
cfg := &config.Config{}
cfg.Secrets.JWTSecret = "test-secret"
+55 -3
View File
@@ -17,10 +17,12 @@ package service
import (
"context"
"crypto/sha1"
"errors"
"fmt"
"math"
"os"
"net/url"
"os"
"path"
"path/filepath"
"regexp"
@@ -821,7 +823,7 @@ func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QB
// (onTorrentComplete 内部仍受 organize.auto 开关约束,且
// 整理对已存在的目标文件幂等跳过)。
d.prevStates[stateKey] = true
if recentlyCompletedTorrent(torrent, time.Now()) {
if recentlyCompletedTorrent(torrent, time.Now()) && !d.completedTorrentCatchupRecorded(ctx, torrent) {
shouldQueue = true
}
case complete && !wasComplete:
@@ -928,6 +930,8 @@ func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent)
// 防止每次启动都把全部历史种子重新过一遍整理流程。
const completedTorrentCatchupWindow = 24 * time.Hour
const completedTorrentCatchupSettingPrefix = "download.auto_organized."
// recentlyCompletedTorrent 报告该种子是否在补整理时间窗内完成。
// qBittorrent 未提供 completion_on 时保守地返回 false。
func recentlyCompletedTorrent(torrent QBitTorrent, now time.Time) bool {
@@ -938,6 +942,46 @@ func recentlyCompletedTorrent(torrent QBitTorrent, now time.Time) bool {
return now.Sub(completed) <= completedTorrentCatchupWindow
}
func (d *DownloadService) completedTorrentCatchupRecorded(ctx context.Context, torrent QBitTorrent) bool {
if d == nil || d.repo == nil || d.repo.Setting == nil {
return false
}
key := completedTorrentCatchupSettingKey(torrent)
if key == "" {
return false
}
value, err := d.repo.Setting.Get(ctx, key)
if err != nil {
return false
}
return parseBoolSetting(value, false)
}
func (d *DownloadService) markCompletedTorrentCatchupRecorded(ctx context.Context, torrent QBitTorrent) {
if d == nil || d.repo == nil || d.repo.Setting == nil {
return
}
key := completedTorrentCatchupSettingKey(torrent)
if key == "" {
return
}
if err := d.repo.Setting.Set(ctx, key, "true"); err != nil && d.log != nil {
d.log.Debug("mark completed torrent catchup failed",
zap.String("hash", torrent.Hash),
zap.String("name", torrent.Name),
zap.Error(err))
}
}
func completedTorrentCatchupSettingKey(torrent QBitTorrent) string {
key := completedTorrentQueueKey(torrent)
if key == "" {
return ""
}
sum := sha1.Sum([]byte(key))
return completedTorrentCatchupSettingPrefix + fmt.Sprintf("%x", sum[:])
}
func completedTorrentQueueKey(torrent QBitTorrent) string {
hash := strings.ToLower(strings.TrimSpace(torrent.Hash))
if hash != "" {
@@ -1054,9 +1098,17 @@ func (d *DownloadService) onTorrentComplete(ctx context.Context, torrent QBitTor
zap.Error(err))
return
}
if d.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" {
if d.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultHasChanges(res) {
res.Scans, res.Scrapes = d.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, d.repo))
} else if d.log != nil && res != nil && !OrganizeResultHasChanges(res) {
d.log.Info("auto organize completed torrent skipped scan; no destination changes",
zap.String("hash", torrent.Hash),
zap.String("source", source),
zap.Int("organized", res.Organized),
zap.Int("replaced", res.Replaced),
zap.Int("skipped", res.Skipped))
}
d.markCompletedTorrentCatchupRecorded(context.Background(), torrent)
d.log.Info("auto organize completed torrent finished",
zap.String("hash", torrent.Hash),
zap.String("source", source),
+69
View File
@@ -206,6 +206,75 @@ func TestDownloadPollCatchesUpRecentlyCompletedTorrents(t *testing.T) {
}
}
func TestDownloadPollSkipsRecordedCompletedTorrentCatchup(t *testing.T) {
repos := newOrganizerTestRepo(t)
torrent := QBitTorrent{
Hash: "fresh-complete",
Name: "Fresh Complete S01E01",
Progress: 1,
CompletionOn: time.Now().Add(-time.Hour).Unix(),
}
if err := repos.Setting.Set(t.Context(), completedTorrentCatchupSettingKey(torrent), "true"); err != nil {
t.Fatal(err)
}
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil)
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{torrent}, nil)
if got := len(svc.organizeQueue); got != 0 {
t.Fatalf("recorded completed torrent queued %d organize jobs, want 0", got)
}
}
func TestAutoOrganizeSkipsScanWhenNoFilesChanged(t *testing.T) {
root := t.TempDir()
src := filepath.Join(root, "downloads", "国产剧", "狂飙.S01E01.2023.1080p.mkv")
dest := filepath.Join(root, "media")
writeOrgFile(t, src, "episode")
repos := newOrganizerTestRepo(t)
for key, value := range map[string]string{
"organizer.auto_after_download": "true",
"organize.target_dir": dest,
"organize.transfer_mode": "copy",
} {
if err := repos.Setting.Set(t.Context(), key, value); err != nil {
t.Fatal(err)
}
}
lib := model.Library{Name: "国产剧", Path: filepath.Join(dest, "电视剧", "国产剧"), Type: "tv", Enabled: true}
if err := repos.Library.Create(t.Context(), &lib); err != nil {
t.Fatal(err)
}
org := NewOrganizerService(&config.Config{}, zap.NewNop(), repos)
if _, err := org.OrganizeDirectory(t.Context(), OrganizeOptions{
SourcePath: src,
DestPath: dest,
TransferMode: TransferCopy,
}); err != nil {
t.Fatalf("seed organized destination: %v", err)
}
scanner := NewScannerService(&config.Config{}, zap.NewNop(), repos, NewHub(zap.NewNop()), nil, nil)
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), org)
svc.SetScanner(scanner)
svc.onTorrentComplete(t.Context(), QBitTorrent{
Hash: "done123",
Name: "狂飙.S01E01.2023.1080p",
Progress: 1,
SavePath: filepath.Dir(src),
ContentPath: src,
})
var count int64
if err := repos.DB.Model(&model.Media{}).Count(&count).Error; err != nil {
t.Fatal(err)
}
if count != 0 {
t.Fatalf("no-op auto organize triggered scan and created %d media rows, want 0", count)
}
}
func TestCompletedTorrentSourceUsesConfiguredMapping(t *testing.T) {
root := t.TempDir()
localRoot := filepath.Join(root, "localdl")
+8
View File
@@ -47,6 +47,14 @@ func OrganizeScrapeAfterEnabled(ctx context.Context, repo *repository.Container)
return false
}
// OrganizeResultHasChanges reports whether an organize run actually changed
// files in the destination library. Skipped duplicates are intentionally not a
// change: scanning after a no-op organize can turn a harmless restart into a
// full library ffprobe sweep.
func OrganizeResultHasChanges(res *OrganizeResult) bool {
return res != nil && (res.Organized > 0 || res.Replaced > 0)
}
// ScanLibrariesForPath recursively scans libraries affected by an organize
// destination. If preferredLibraryID is set, only that library is scanned.
// Otherwise every enabled library whose path intersects destRoot is scanned;
+9 -2
View File
@@ -18,9 +18,16 @@ func translateClientPath(clientPath string, mappings map[string]string) string {
return clean
}
// 尝试路径映射
cleanForMatch := filepath.ToSlash(clean)
for clientPrefix, localPrefix := range mappings {
if strings.HasPrefix(clean, clientPrefix) {
translated := filepath.Join(localPrefix, strings.TrimPrefix(clean, clientPrefix))
prefix := strings.TrimRight(filepath.ToSlash(filepath.Clean(clientPrefix)), "/")
if prefix == "" || prefix == "." {
continue
}
if cleanForMatch == prefix || strings.HasPrefix(cleanForMatch, prefix+"/") {
rel := strings.TrimPrefix(cleanForMatch, prefix)
rel = strings.TrimPrefix(rel, "/")
translated := filepath.Join(localPrefix, filepath.FromSlash(rel))
if _, err := os.Stat(translated); err == nil {
return translated
}
+13 -5
View File
@@ -135,10 +135,10 @@ func (s *SchedulerService) Start(ctx context.Context) {
}
for _, j := range s.jobs {
initialDelay := 15 * time.Second
if j.name == "library_scan" {
// 重启后不立即整库重扫:更新/重启窗口恰是登录高峰,启动
// 15 秒即全量扫描曾把 CPU/磁盘打满导致无法登录。首轮等满
// 一个完整周期再跑,平时的每小时节奏不变。
if j.name == "library_scan" || j.name == "organize_source" {
// 重启后不立即整库重扫/整理下载目录:更新窗口恰是登录高峰,
// 15 秒即全量 walk + ffprobe 曾把 CPU/磁盘打满导致无法登录。
// 首轮等满一个完整周期再跑,平时节奏不变。
initialDelay = j.interval
}
go s.loopWithInitialDelay(ctx, j, initialDelay)
@@ -483,8 +483,16 @@ func (s *SchedulerService) jobOrganizeSource(ctx context.Context) error {
if err != nil {
return err
}
if s.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" {
if s.scanner != nil && res != nil && strings.TrimSpace(res.DestPath) != "" && OrganizeResultHasChanges(res) {
res.Scans, res.Scrapes = s.scanner.ScanAndScrapeLibrariesForPath(ctx, res.DestPath, "", OrganizeScrapeAfterEnabled(ctx, s.repo))
} else if s.log != nil && res != nil && !OrganizeResultHasChanges(res) {
s.log.Info("scheduled source organize skipped scan; no destination changes",
zap.String("source", res.SourcePath),
zap.String("dest", res.DestPath),
zap.Int("organized", res.Organized),
zap.Int("replaced", res.Replaced),
zap.Int("skipped", res.Skipped),
)
}
if s.log != nil && res != nil {
s.log.Info("scheduled source organize finished",
+63 -3
View File
@@ -5,6 +5,7 @@ package service
import (
"context"
"strconv"
"strings"
"time"
@@ -268,15 +269,21 @@ func (c *Container) warmMediaSearchIndex(ctx context.Context) {
if c == nil || c.Repo == nil || c.Repo.Media == nil {
return
}
if !mediaSearchWarmupEnabled(ctx, c.Repo) {
if c.Log != nil {
c.Log.Info("media search index warmup disabled")
}
return
}
// 错峰:FTS 正常由 media 表触发器实时维护,回填只是升级或异常后的
// 兜底。先让登录、首页等关键路径跑起来,再开始后台补索引。
select {
case <-ctx.Done():
return
case <-time.After(30 * time.Second):
case <-time.After(mediaSearchWarmupDelay(ctx, c.Repo)):
}
const batchSize = 1000
const pause = 100 * time.Millisecond
batchSize := mediaSearchWarmupBatchSize(ctx, c.Repo)
pause := mediaSearchWarmupPause(ctx, c.Repo)
total := int64(0)
for {
select {
@@ -304,6 +311,59 @@ func (c *Container) warmMediaSearchIndex(ctx context.Context) {
}
}
func mediaSearchWarmupEnabled(ctx context.Context, repo *repository.Container) bool {
if repo == nil || repo.Setting == nil {
return true
}
value, err := repo.Setting.Get(ctx, "search.index_warmup_enabled")
if err != nil || strings.TrimSpace(value) == "" {
return true
}
return parseBoolSetting(value, true)
}
func mediaSearchWarmupDelay(ctx context.Context, repo *repository.Container) time.Duration {
seconds := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_delay_seconds", 120)
if seconds < 30 {
seconds = 30
}
return time.Duration(seconds) * time.Second
}
func mediaSearchWarmupBatchSize(ctx context.Context, repo *repository.Container) int {
size := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_batch_size", 100)
if size < 10 {
size = 10
}
if size > 1000 {
size = 1000
}
return size
}
func mediaSearchWarmupPause(ctx context.Context, repo *repository.Container) time.Duration {
ms := mediaSearchWarmupIntSetting(ctx, repo, "search.index_warmup_pause_ms", 2000)
if ms < 250 {
ms = 250
}
return time.Duration(ms) * time.Millisecond
}
func mediaSearchWarmupIntSetting(ctx context.Context, repo *repository.Container, key string, fallback int) int {
if repo == nil || repo.Setting == nil {
return fallback
}
value, err := repo.Setting.Get(ctx, key)
if err != nil {
return fallback
}
n, err := strconv.Atoi(strings.TrimSpace(value))
if err != nil || n <= 0 {
return fallback
}
return n
}
func (c *Container) NormalizeCloudLibraryTypes(ctx context.Context) error {
if c == nil || c.Repo == nil || c.Repo.Library == nil || c.Repo.DB == nil {
return nil
+61 -19
View File
@@ -107,25 +107,17 @@ func (s *TokenService) issuePair(ctx context.Context, userID, role, tier string,
TokenHash: tokenHash,
ExpiresAt: time.Now().Add(RefreshTokenDuration),
}
storeCtx := ctx
cancel := func() {}
if bestEffort {
storeCtx, cancel = context.WithTimeout(context.Background(), loginRefreshTokenStoreTimeout)
s.storeRefreshTokenBestEffort(userID, tokenHash, rt.ExpiresAt)
return &TokenPair{
AccessToken: accessToken,
RefreshToken: refreshToken,
ExpiresIn: int64(AccessTokenDuration.Seconds()),
TokenType: "Bearer",
}, nil
}
err = s.storeRefreshToken(storeCtx, rt)
cancel()
if err != nil {
if !bestEffort {
return nil, err
}
if s.log != nil {
s.log.Warn("refresh token store delayed; login will continue",
zap.String("user_id", userID),
zap.Error(err))
}
if s.trackDelayedStore(userID, tokenHash, rt.ExpiresAt) {
go s.storeRefreshTokenEventually(userID, tokenHash, rt.ExpiresAt)
}
if err := s.storeRefreshToken(ctx, rt); err != nil {
return nil, err
}
return &TokenPair{
@@ -136,6 +128,56 @@ func (s *TokenService) issuePair(ctx context.Context, userID, role, tier string,
}, nil
}
func (s *TokenService) storeRefreshTokenBestEffort(userID, tokenHash string, expiresAt time.Time) {
if !s.trackDelayedStore(userID, tokenHash, expiresAt) {
return
}
done := make(chan error, 1)
go func() {
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
defer cancel()
done <- s.storeRefreshToken(ctx, &model.RefreshToken{
UserID: userID,
TokenHash: tokenHash,
ExpiresAt: expiresAt,
})
}()
select {
case err := <-done:
s.finishBestEffortRefreshTokenStore(userID, tokenHash, expiresAt, err)
case <-time.After(loginRefreshTokenStoreTimeout):
if s.log != nil {
s.log.Warn("refresh token store delayed; login will continue",
zap.String("user_id", userID),
zap.Error(context.DeadlineExceeded))
}
go func() {
err := <-done
s.finishBestEffortRefreshTokenStore(userID, tokenHash, expiresAt, err)
}()
}
}
func (s *TokenService) finishBestEffortRefreshTokenStore(userID, tokenHash string, expiresAt time.Time, err error) {
if err == nil {
s.untrackDelayedStore(userID, tokenHash)
return
}
if repository.IsSQLiteBusyError(err) || errors.Is(err, context.DeadlineExceeded) || errors.Is(err, context.Canceled) {
if s.log != nil {
s.log.Warn("refresh token store delayed; login will continue",
zap.String("user_id", userID),
zap.Error(err))
}
s.storeRefreshTokenEventually(userID, tokenHash, expiresAt)
return
}
s.untrackDelayedStore(userID, tokenHash)
if s.log != nil {
s.log.Warn("refresh token delayed store failed permanently", zap.String("user_id", userID), zap.Error(err))
}
}
func (s *TokenService) storeRefreshToken(ctx context.Context, rt *model.RefreshToken) error {
if err := s.repo.RefreshToken.Create(ctx, rt); err != nil {
return err
@@ -148,7 +190,7 @@ func (s *TokenService) storeRefreshToken(ctx context.Context, rt *model.RefreshT
func (s *TokenService) storeRefreshTokenEventually(userID, tokenHash string, expiresAt time.Time) {
defer s.untrackDelayedStore(userID, tokenHash)
delay := 5 * time.Second
delay := time.Second
for attempt := 1; attempt <= 8; attempt++ {
timer := time.NewTimer(delay)
<-timer.C
@@ -313,7 +355,7 @@ func (s *TokenService) Refresh(ctx context.Context, refreshToken string) (*Token
s.untrackDelayedStore(rt.UserID, tokenHash)
// 签发新的令牌对
return s.IssuePair(ctx, user.ID, user.Role, user.Tier)
return s.IssuePairBestEffort(ctx, user.ID, user.Role, user.Tier)
}
// RevokeAll 撤销用户的所有 Refresh Token(用于登出)。