fix: keep login responsive during background organize

This commit is contained in:
ShukeBta
2026-06-11 19:28:49 +08:00
parent 8b32d99b5c
commit d9f87c61b6
7 changed files with 260 additions and 34 deletions
+2 -2
View File
@@ -29,8 +29,8 @@ database:
wal_mode: true
busy_timeout: 5000
cache_size: -20000 # KB (negative)
max_open_conns: 1
max_idle_conns: 1
max_open_conns: 4 # small WAL pool; keeps login/playback responsive during scans
max_idle_conns: 2
secrets:
# When empty, a 32-byte random secret is generated on first start and
+16 -6
View File
@@ -21,6 +21,11 @@ import (
// EnvPrefix 是所有环境变量驱动的覆盖使用的前缀。
const EnvPrefix = "MEDIASTATION"
const (
defaultDatabaseMaxOpenConns = 4
defaultDatabaseMaxIdleConns = 2
)
// Config 是根配置聚合。
type Config struct {
App AppConfig `mapstructure:"app"`
@@ -226,8 +231,8 @@ func setDefaults(v *viper.Viper) {
v.SetDefault("database.wal_mode", true)
v.SetDefault("database.busy_timeout", 5000)
v.SetDefault("database.cache_size", -20000)
v.SetDefault("database.max_open_conns", 1)
v.SetDefault("database.max_idle_conns", 1)
v.SetDefault("database.max_open_conns", defaultDatabaseMaxOpenConns)
v.SetDefault("database.max_idle_conns", defaultDatabaseMaxIdleConns)
v.SetDefault("secrets.jwt_secret", "")
@@ -303,6 +308,15 @@ func (c *Config) normalize() error {
if c.Database.DBPath == "" {
c.Database.DBPath = filepath.Join(c.App.DataDir, "mediastation.db")
}
if c.Database.MaxOpenConns <= 1 {
c.Database.MaxOpenConns = defaultDatabaseMaxOpenConns
}
if c.Database.MaxIdleConns <= 0 || c.Database.MaxIdleConns > c.Database.MaxOpenConns {
c.Database.MaxIdleConns = defaultDatabaseMaxIdleConns
if c.Database.MaxIdleConns > c.Database.MaxOpenConns {
c.Database.MaxIdleConns = c.Database.MaxOpenConns
}
}
if c.Cache.CacheDir == "" {
c.Cache.CacheDir = filepath.Join(c.App.DataDir, "cache")
}
@@ -335,7 +349,3 @@ func asConfigFileNotFound(err error, target *viper.ConfigFileNotFoundError) bool
}
return false
}
+26
View File
@@ -28,6 +28,12 @@ func TestLoadDefaults(t *testing.T) {
if cfg.Database.DBPath == "" {
t.Fatalf("expected non-empty DBPath")
}
if cfg.Database.MaxOpenConns != defaultDatabaseMaxOpenConns {
t.Fatalf("expected default MaxOpenConns %d, got %d", defaultDatabaseMaxOpenConns, cfg.Database.MaxOpenConns)
}
if cfg.Database.MaxIdleConns != defaultDatabaseMaxIdleConns {
t.Fatalf("expected default MaxIdleConns %d, got %d", defaultDatabaseMaxIdleConns, cfg.Database.MaxIdleConns)
}
if cfg.Secrets.JWTSecret == "" {
t.Fatalf("expected auto-generated JWT secret")
}
@@ -62,3 +68,23 @@ func TestEnvOverride(t *testing.T) {
t.Fatalf("expected port 9090 from env, got %d", cfg.App.Port)
}
}
func TestLoadHealsHistoricalSingleConnectionDatabaseConfig(t *testing.T) {
dir := t.TempDir()
wd, _ := os.Getwd()
defer func() { _ = os.Chdir(wd) }()
if err := os.Chdir(dir); err != nil {
t.Fatalf("chdir: %v", err)
}
if err := os.WriteFile("config.yaml", []byte("database:\n max_open_conns: 1\n max_idle_conns: 1\n"), 0o644); err != nil {
t.Fatal(err)
}
cfg, err := Load()
if err != nil {
t.Fatalf("Load() error: %v", err)
}
if cfg.Database.MaxOpenConns != defaultDatabaseMaxOpenConns {
t.Fatalf("expected historical MaxOpenConns=1 to heal to %d, got %d", defaultDatabaseMaxOpenConns, cfg.Database.MaxOpenConns)
}
}
+2
View File
@@ -86,6 +86,8 @@ func AutoMigrate(db *gorm.DB) error {
func ensurePerformanceIndexes(db *gorm.DB) error {
statements := []string{
`CREATE INDEX IF NOT EXISTS idx_media_created_active ON media(created_at DESC) WHERE deleted_at IS NULL`,
`CREATE INDEX IF NOT EXISTS idx_media_episode_created_active ON media(created_at DESC) WHERE deleted_at IS NULL AND (season_num > 0 OR episode_num > 0)`,
`CREATE INDEX IF NOT EXISTS idx_media_library_created_active ON media(library_id, created_at DESC) WHERE deleted_at IS NULL`,
`CREATE INDEX IF NOT EXISTS idx_media_library_episode_active ON media(library_id, season_num, episode_num, created_at DESC) WHERE deleted_at IS NULL`,
`CREATE INDEX IF NOT EXISTS idx_media_series_active ON media(series_id, season_num, episode_num) WHERE deleted_at IS NULL`,
+2
View File
@@ -58,6 +58,8 @@ func TestEnsurePerformanceIndexesCreatesHotPathIndexes(t *testing.T) {
t.Fatal(err)
}
for _, name := range []string{
"idx_media_created_active",
"idx_media_episode_created_active",
"idx_media_library_created_active",
"idx_media_library_episode_active",
"idx_favorites_user_media_active",
+170 -26
View File
@@ -44,10 +44,14 @@ type DownloadService struct {
scanner *ScannerService
site *SiteService
mu sync.Mutex
stopCh chan struct{}
pollOnce sync.Once
prevStates map[string]bool // hash -> wasCompleted
mu sync.Mutex
stopCh chan struct{}
pollOnce sync.Once
organizeOnce sync.Once
prevStates map[string]bool // hash -> wasCompleted
pollInitialized bool
organizeQueue chan QBitTorrent
organizeQueued map[string]struct{}
}
func (d *DownloadService) SetScanner(scanner *ScannerService) {
@@ -58,6 +62,10 @@ var torrentEpisodeToken = regexp.MustCompile(`(?i)e\d{1,3}`)
const settingDownloadClientsManaged = "download_clients.managed"
const completedTorrentOrganizeQueueSize = 64
var completedTorrentOrganizeCooldown = 3 * time.Second
// ErrDownloadAlreadyExists tells callers that the requested resource is already
// tracked locally or present in qBittorrent. Subscriptions treat this as a
// successful dedup hit, not as a retryable enqueue failure.
@@ -132,14 +140,16 @@ func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub, o
siteSvc = site[0]
}
return &DownloadService{
log: log,
repo: repo,
hub: hub,
qb: NewQBitClient(log, QBitConfig{}),
organizer: organizer,
site: siteSvc,
prevStates: make(map[string]bool),
stopCh: make(chan struct{}),
log: log,
repo: repo,
hub: hub,
qb: NewQBitClient(log, QBitConfig{}),
organizer: organizer,
site: siteSvc,
prevStates: make(map[string]bool),
organizeQueue: make(chan QBitTorrent, completedTorrentOrganizeQueueSize),
organizeQueued: make(map[string]struct{}),
stopCh: make(chan struct{}),
}
}
@@ -147,6 +157,7 @@ func NewDownloadService(log *zap.Logger, repo *repository.Container, hub *Hub, o
func (d *DownloadService) Start(ctx context.Context) {
d.pollOnce.Do(func() {
_ = d.ReloadConfig(ctx)
d.startAutoOrganizeWorker(ctx)
go d.poll(ctx)
})
}
@@ -709,7 +720,11 @@ func (d *DownloadService) Delete(ctx context.Context, hash string, withFiles boo
return err
}
d.markDownloadTaskDeleted(ctx, torrentName)
delete(d.prevStates, hash)
stateKey := strings.ToLower(hash)
d.mu.Lock()
delete(d.prevStates, stateKey)
delete(d.organizeQueued, stateKey)
d.mu.Unlock()
return nil
}
@@ -770,21 +785,152 @@ func (d *DownloadService) poll(ctx context.Context) {
}
rows, _ := d.repo.Download.List(ctx)
taskByKey := tasksByIdentity(rows)
// Detect completed downloads and trigger organize
for _, t := range live {
hash := t.Hash
complete := t.Progress >= 1.0
d.syncDownloadTaskProgress(ctx, t, taskByKey)
if complete && !d.prevStates[hash] {
// Just completed: trigger organize
go d.onTorrentComplete(ctx, t)
}
d.prevStates[hash] = complete
}
d.processDownloadSnapshot(ctx, live, taskByKey)
d.hub.Publish("download", map[string]any{"torrents": live})
}
}
func (d *DownloadService) processDownloadSnapshot(ctx context.Context, live []QBitTorrent, taskByKey map[string]model.DownloadTask) {
d.mu.Lock()
if d.prevStates == nil {
d.prevStates = make(map[string]bool)
}
firstSnapshot := !d.pollInitialized
if firstSnapshot {
d.pollInitialized = true
}
d.mu.Unlock()
for _, torrent := range live {
stateKey := completedTorrentQueueKey(torrent)
complete := torrent.Progress >= 1.0
d.syncDownloadTaskProgress(ctx, torrent, taskByKey)
if stateKey == "" {
continue
}
shouldQueue := false
d.mu.Lock()
wasComplete, wasSeen := d.prevStates[stateKey]
switch {
case complete && (firstSnapshot || !wasSeen):
d.prevStates[stateKey] = true
case complete && !wasComplete:
shouldQueue = true
case complete:
d.prevStates[stateKey] = true
default:
d.prevStates[stateKey] = false
}
d.mu.Unlock()
if shouldQueue && d.enqueueCompletedTorrent(torrent) {
d.mu.Lock()
d.prevStates[stateKey] = true
d.mu.Unlock()
}
}
}
func (d *DownloadService) startAutoOrganizeWorker(ctx context.Context) {
d.mu.Lock()
if d.organizeQueue == nil {
d.organizeQueue = make(chan QBitTorrent, completedTorrentOrganizeQueueSize)
}
if d.organizeQueued == nil {
d.organizeQueued = make(map[string]struct{})
}
d.mu.Unlock()
d.organizeOnce.Do(func() {
go d.autoOrganizeWorker(ctx)
})
}
func (d *DownloadService) enqueueCompletedTorrent(torrent QBitTorrent) bool {
key := completedTorrentQueueKey(torrent)
if key == "" {
return false
}
d.mu.Lock()
if d.organizeQueue == nil {
d.organizeQueue = make(chan QBitTorrent, completedTorrentOrganizeQueueSize)
}
if d.organizeQueued == nil {
d.organizeQueued = make(map[string]struct{})
}
if _, ok := d.organizeQueued[key]; ok {
d.mu.Unlock()
return true
}
select {
case d.organizeQueue <- torrent:
d.organizeQueued[key] = struct{}{}
d.mu.Unlock()
return true
default:
d.mu.Unlock()
if d.log != nil {
d.log.Warn("auto organize queue full; will retry completed torrent later",
zap.String("hash", torrent.Hash),
zap.String("name", torrent.Name))
}
return false
}
}
func (d *DownloadService) autoOrganizeWorker(ctx context.Context) {
for {
select {
case <-ctx.Done():
return
case <-d.stopCh:
return
case torrent := <-d.organizeQueue:
d.onTorrentComplete(ctx, torrent)
d.markCompletedTorrentOrganizeDone(torrent)
if completedTorrentOrganizeCooldown <= 0 {
continue
}
timer := time.NewTimer(completedTorrentOrganizeCooldown)
select {
case <-ctx.Done():
timer.Stop()
return
case <-d.stopCh:
timer.Stop()
return
case <-timer.C:
}
}
}
}
func (d *DownloadService) markCompletedTorrentOrganizeDone(torrent QBitTorrent) {
key := completedTorrentQueueKey(torrent)
if key == "" {
return
}
d.mu.Lock()
delete(d.organizeQueued, key)
d.mu.Unlock()
}
func completedTorrentQueueKey(torrent QBitTorrent) string {
hash := strings.ToLower(strings.TrimSpace(torrent.Hash))
if hash != "" {
return hash
}
parts := []string{torrent.Name, torrent.ContentPath, torrent.SavePath}
for i := range parts {
parts[i] = strings.TrimSpace(parts[i])
}
key := strings.Join(parts, "|")
if strings.Trim(key, "|") == "" {
return ""
}
return strings.ToLower(key)
}
func (d *DownloadService) syncDownloadTaskProgress(ctx context.Context, torrent QBitTorrent, taskByKey map[string]model.DownloadTask) {
if d == nil || d.repo == nil || d.repo.DB == nil || strings.TrimSpace(torrent.Name) == "" {
return
@@ -916,5 +1062,3 @@ func (d *DownloadService) completedTorrentSource(torrent QBitTorrent) string {
}
return ""
}
+42
View File
@@ -87,6 +87,48 @@ func TestDownloadCompleteAutoOrganizesContentPath(t *testing.T) {
}
}
func TestDownloadPollBaselinesAlreadyCompletedTorrents(t *testing.T) {
repos := newOrganizerTestRepo(t)
svc := NewDownloadService(zap.NewNop(), repos, NewHub(zap.NewNop()), nil)
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{
Hash: "already-complete",
Name: "Already Complete S01E01",
Progress: 1,
}}, nil)
if got := len(svc.organizeQueue); got != 0 {
t.Fatalf("first poll queued %d organize jobs, want 0", got)
}
if !svc.prevStates["already-complete"] {
t.Fatal("first poll should remember completed baseline state")
}
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{
Hash: "late-complete",
Name: "Late Complete S01E01",
Progress: 1,
}}, nil)
if got := len(svc.organizeQueue); got != 0 {
t.Fatalf("newly discovered completed torrent queued %d organize jobs, want 0", got)
}
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{
Hash: "new-download",
Name: "New Download S01E01",
Progress: 0.5,
}}, nil)
svc.processDownloadSnapshot(t.Context(), []QBitTorrent{{
Hash: "new-download",
Name: "New Download S01E01",
Progress: 1,
}}, nil)
if got := len(svc.organizeQueue); got != 1 {
t.Fatalf("completion transition queued %d organize jobs, want 1", got)
}
}
func TestPublicDownloadTitleUsesMagnetDisplayName(t *testing.T) {
got := publicDownloadTitle("magnet:?xt=urn:btih:abc&dn=%E6%B5%8B%E8%AF%95%E5%BD%B1%E7%89%87")
if got != "测试影片" {