diff --git a/README.md b/README.md index 60bf959..c9d5e72 100644 --- a/README.md +++ b/README.md @@ -105,3 +105,19 @@ go build -o am-downloader . 升级前仍建议先备份 `data/downloader.db`。Docker 部署时不要把 `data/` 或 `downloads/` 加入 Git。 当前版本不会主动删除旧媒体文件;媒体资源记录缺失本地文件时会在下一次请求中重新取得资源。 + +## 多用户并发与恢复 + +下载任务和共享媒体资源分离: + +- 同一 source + rid + quality 只允许一个 Asset 物理下载,多个用户 Task 可以共享。 +- 每个用户 Task 保留独立 WebDAV 目标路径、状态和额度预留。 +- Task 使用数据库条件更新抢占状态,重复点击不会启动多个 worker。 +- Asset 使用数据库租约(heartbeat/lease)防止 worker 崩溃后永久占锁;过期租约会自动恢复旧任务。 +- 下载文件使用按 Asset/generation 隔离的缓存目录和 task-specific .part 临时文件,避免并发用户覆盖同一个本地文件。 +- WebDAV 上传采用独立的 upload_pending -> uploading -> done 状态机;上传失败只重试上传,不重新扣积分。 +- 已提交额度的历史失败任务会自动恢复;已退款任务不会因为后台 resync 被无条件重新扣费。 +- SQLite 使用 WAL + busy timeout,并对关键事务做锁冲突重试;高并发写入仍由 SQLite 单写者模型串行化。 +- 本地共享文件只有在没有其他活动 Task 需要时才允许清理。 + +并发回归测试覆盖:Task 抢占、共享 Asset 租约恢复、已提交额度重试、已退款额度重试、并发积分预留以及 Asset generation 隔离。 diff --git a/assets.go b/assets.go index 2074be5..29bae49 100644 --- a/assets.go +++ b/assets.go @@ -2,9 +2,6 @@ package main import ( "errors" - "fmt" - "os" - "path/filepath" "strings" "time" @@ -23,6 +20,8 @@ type MediaAsset struct { Size int64 `json:"size"` Status string `gorm:"index;size:20" json:"status"` OwnerTaskID uint `gorm:"index" json:"ownerTaskId,omitempty"` + LeaseUntil time.Time `gorm:"index" json:"-"` + Generation uint64 `json:"generation"` CreatedAt time.Time `json:"createdAt"` UpdatedAt time.Time `json:"updatedAt"` } @@ -58,70 +57,9 @@ func (a *App) ensureAsset(t *Task) (*MediaAsset, error) { return &asset, nil } -// acquireAsset serializes physical downloads for the same source/rid/quality. -// A waiting task yields its worker briefly instead of creating duplicate files. -func (a *App) acquireAsset(t *Task) (*MediaAsset, bool, error) { - for i := 0; i < 120; i++ { - asset, err := a.ensureAsset(t) - if err != nil { - return nil, false, err - } - if asset.Status == "done" && asset.LocalPath != "" { - if fi, statErr := os.Stat(asset.LocalPath); statErr == nil && fi.Size() > 0 { - t.LocalPath, t.Format, t.ActualQuality = asset.LocalPath, asset.Format, asset.ActualQuality - t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = fi.Size(), fi.Size(), 100, "done" - if err := a.db.Save(t).Error; err != nil { - return nil, false, err - } - return asset, true, nil - } - } - claimed := false - err = a.db.Transaction(func(tx *gorm.DB) error { - var current MediaAsset - if err := tx.First(¤t, asset.ID).Error; err != nil { - return err - } - if current.Status == "downloading" && current.OwnerTaskID != t.ID { - if current.UpdatedAt.After(time.Now().Add(-10 * time.Minute)) { - return nil - } - } - current.Status, current.OwnerTaskID = "downloading", t.ID - if current.LocalPath == "" { - current.LocalPath = filepath.Join(a.cfg.DownloadDir, safe(t.Artist)+" - "+safe(t.Title)+"."+t.Format) - } - if err := tx.Save(¤t).Error; err != nil { - return err - } - *asset = current - claimed = true - return nil - }) - if err != nil { - return nil, false, err - } - if claimed { - t.LocalPath = asset.LocalPath - return asset, false, nil - } - time.Sleep(500 * time.Millisecond) - } - return nil, false, fmt.Errorf("timed out waiting for shared media asset") -} +// acquireAsset is the compatibility entry point. +func (a *App) acquireAsset(t *Task) (*MediaAsset, bool, error) { return a.claimAsset(t) } -func (a *App) markAssetDone(t *Task) { - if t.AssetID == 0 { - return - } - _ = a.db.Model(&MediaAsset{}).Where("id = ? AND owner_task_id = ?", t.AssetID, t.ID).Updates(map[string]any{ - "status": "done", "owner_task_id": 0, "local_path": t.LocalPath, "format": t.Format, "actual_quality": t.ActualQuality, "size": t.DownloadedBytes, - }).Error -} +func (a *App) markAssetDone(t *Task) { _ = a.finishAsset(t) } -func (a *App) releaseAsset(t *Task) { - if t.AssetID == 0 { - return - } - _ = a.db.Model(&MediaAsset{}).Where("id = ? AND owner_task_id = ?", t.AssetID, t.ID).Updates(map[string]any{"status": "error", "owner_task_id": 0}).Error -} +func (a *App) releaseAsset(t *Task) { a.invalidateAsset(t) } diff --git a/concurrency.go b/concurrency.go new file mode 100644 index 0000000..a8a757f --- /dev/null +++ b/concurrency.go @@ -0,0 +1,255 @@ +package main + +import ( + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "os" + "path/filepath" + "time" + + "gorm.io/gorm" +) + +const assetLease = 15 * time.Minute + +var ( + errTaskRunning = errors.New("task is already running") + errTaskAlreadyDone = errors.New("task is already completed") +) + +func (a *App) claimTask(id uint) (*Task, error) { + result := a.db.Model(&Task{}).Where("id = ? AND status = ?", id, "pending"). + Updates(map[string]any{"status": "resolving", "error": "", "progress": 0}) + if result.Error != nil { + return nil, result.Error + } + if result.RowsAffected != 1 { + var t Task + if err := a.db.First(&t, id).Error; err != nil { + return nil, err + } + if t.Status == "upload_pending" { + return nil, fmt.Errorf("task %d is waiting for upload", id) + } + if t.Status == "done" { + return nil, errTaskAlreadyDone + } + return nil, errTaskRunning + } + var t Task + if err := a.db.First(&t, id).Error; err != nil { + return nil, err + } + return &t, nil +} + +func (a *App) claimUpload(t *Task) bool { + result := a.db.Model(&Task{}). + Where("id = ? AND (status = ? OR (status = ? AND (remote_path = '' OR remote_path IS NULL)))", + t.ID, "upload_pending", "done"). + Updates(map[string]any{"status": "uploading", "error": ""}) + if result.Error != nil || result.RowsAffected != 1 { + return false + } + t.Status, t.Error = "uploading", "" + return true +} + +func (a *App) taskReservationStatus(taskID, reservationID uint) (string, error) { + if reservationID == 0 { + return "", nil + } + var r CreditReservation + if err := a.db.First(&r, reservationID).Error; err != nil { + return "", err + } + if r.TaskID != taskID { + return "", fmt.Errorf("credit reservation does not belong to task") + } + return r.Status, nil +} + +func (a *App) repairCommittedTask(t *Task) error { + status, err := a.taskReservationStatus(t.ID, t.CreditReservationID) + if err != nil { + return err + } + if status == "committed" { + if t.Status == "error" { + t.Status, t.Error = "upload_pending", "下载已完成,正在恢复 WebDAV 上传" + return a.db.Model(t).Updates(map[string]any{ + "status": t.Status, "error": t.Error, + }).Error + } + return nil + } + return nil +} + +func (a *App) markAssetHeartbeat(assetID, taskID uint) { + if assetID == 0 || taskID == 0 { + return + } + now := time.Now() + lease := now.Add(assetLease) + _ = a.db.Model(&MediaAsset{}). + Where("id = ? AND owner_task_id = ? AND status = ?", assetID, taskID, "downloading"). + Updates(map[string]any{"lease_until": lease, "updated_at": now}).Error +} + +func (a *App) assetLeaseExpired(asset *MediaAsset) bool { + return asset.Status == "downloading" && !asset.LeaseUntil.IsZero() && time.Now().After(asset.LeaseUntil) +} +func (a *App) claimAsset(t *Task) (*MediaAsset, bool, error) { + for i := 0; i < 180; i++ { + asset, err := a.ensureAsset(t) + if err != nil { + return nil, false, err + } + if asset.Status == "done" && asset.LocalPath != "" { + if fi, statErr := os.Stat(asset.LocalPath); statErr == nil && fi.Size() > 0 { + t.LocalPath, t.Format, t.ActualQuality = asset.LocalPath, asset.Format, asset.ActualQuality + t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = fi.Size(), fi.Size(), 100, "done" + if err := a.db.Save(t).Error; err != nil { + return nil, false, err + } + return asset, true, nil + } + } + + claimed := false + now := time.Now() + lease := now.Add(assetLease) + err = a.transactionRetry(func(tx *gorm.DB) error { + var current MediaAsset + if err := tx.First(¤t, asset.ID).Error; err != nil { + return err + } + if current.Status == "downloading" && current.OwnerTaskID != t.ID { + if !current.LeaseUntil.IsZero() && now.Before(current.LeaseUntil) { + return nil + } + if current.OwnerTaskID != 0 { + _ = tx.Model(&Task{}).Where("id = ? AND status IN ?", current.OwnerTaskID, []string{"resolving", "downloading"}).Updates(map[string]any{"status": "pending", "error": "共享资源下载锁超时,已自动恢复", "progress": 0}) + } + } + wasEvicted := current.Status == "evicted" + wasStolen := current.Status == "downloading" && current.OwnerTaskID != 0 && current.OwnerTaskID != t.ID && (current.LeaseUntil.IsZero() || !now.Before(current.LeaseUntil)) + current.Status = "downloading" + current.OwnerTaskID = t.ID + current.LeaseUntil = lease + current.Generation++ + if current.LocalPath == "" || wasEvicted || wasStolen { + current.LocalPath = assetCachePath(a.cfg.DownloadDir, current.AssetKey, t.Format, current.Generation) + } + if err := tx.Save(¤t).Error; err != nil { + return err + } + *asset = current + claimed = true + return nil + }) + if err != nil { + return nil, false, err + } + if claimed { + t.LocalPath = asset.LocalPath + return asset, false, nil + } + time.Sleep(500 * time.Millisecond) + } + return nil, false, fmt.Errorf("timed out waiting for shared media asset") +} +func assetCachePath(root, key, format string, generation uint64) string { + sum := sha256.Sum256([]byte(key)) + return filepath.Join(root, ".assets", hex.EncodeToString(sum[:]), fmt.Sprintf("asset-%d.%s", generation, normalizedFormat(format))) +} + +func (a *App) assertAssetOwner(t *Task) bool { + if t.AssetID == 0 { + return true + } + var asset MediaAsset + if a.db.First(&asset, t.AssetID).Error != nil { + return false + } + return asset.OwnerTaskID == t.ID && asset.Status == "downloading" +} + +func (a *App) finishAsset(t *Task) bool { + if t.AssetID == 0 { + return true + } + result := a.db.Model(&MediaAsset{}). + Where("id = ? AND owner_task_id = ? AND status = ?", t.AssetID, t.ID, "downloading"). + Updates(map[string]any{ + "status": "done", "owner_task_id": 0, "lease_until": nil, + "local_path": t.LocalPath, "format": t.Format, + "actual_quality": t.ActualQuality, "size": t.DownloadedBytes, + }) + return result.Error == nil && result.RowsAffected == 1 +} + +func (a *App) invalidateAsset(t *Task) { + if t.AssetID == 0 { + return + } + _ = a.db.Model(&MediaAsset{}). + Where("id = ? AND owner_task_id = ?", t.AssetID, t.ID). + Updates(map[string]any{"status": "error", "owner_task_id": 0, "lease_until": nil}) +} + +func (a *App) prepareRetry(t *Task) error { + status, err := a.taskReservationStatus(t.ID, t.CreditReservationID) + if err != nil { + return err + } + switch status { + case "committed": + t.Status = "upload_pending" + t.Error = "下载已完成,恢复 WebDAV 上传" + return a.db.Model(t).Updates(map[string]any{ + "status": "upload_pending", "error": t.Error, + }).Error + case "reserved": + t.Status, t.Error = "pending", "" + return a.db.Model(t).Updates(map[string]any{ + "status": "pending", "error": "", "progress": 0, + }).Error + case "refunded", "": + r, reserveErr := a.reserveCredit(t.UserID, t.ID, 1) + if reserveErr != nil { + return reserveErr + } + t.CreditReservationID = r.ID + t.Status, t.Error, t.Progress, t.DownloadedBytes = "pending", "", 0, 0 + return a.db.Model(t).Updates(map[string]any{ + "status": "pending", "error": "", "progress": 0, + "downloaded_bytes": 0, "credit_reservation_id": r.ID, + }).Error + default: + return errReservationState + } +} + +func (a *App) enqueueTask(id uint) { + select { + case a.queue <- id: + default: + go func() { a.queue <- id }() + } +} + +func (a *App) transactionRetry(fn func(*gorm.DB) error) error { + var err error + for attempt := 0; attempt < 8; attempt++ { + err = a.db.Transaction(fn) + if !isSQLiteBusy(err) { + return err + } + time.Sleep(time.Duration(25*(1< 0 && n != t.TotalBytes { a.fail(t, fmt.Errorf("incomplete download: got %d of %d bytes", n, t.TotalBytes)) + _ = os.Remove(tmp) + return + } + if !a.assertAssetOwner(t) { + _ = os.Remove(tmp) return } @@ -131,26 +195,40 @@ func (a *App) downloadFile(t *Task, info playInfo) { } actual = normalizedFormat(actual) finalPath := strings.TrimSuffix(t.LocalPath, filepath.Ext(t.LocalPath)) + "." + actual - if err = os.Rename(tmp, finalPath); err != nil { + if err := os.Rename(tmp, finalPath); err != nil { a.fail(t, err) _ = os.Remove(tmp) return } - t.Format, t.LocalPath, t.ActualQuality, t.DownloadedBytes = actual, finalPath, actualQuality(t.Quality, t.ResolvedQuality, actual), n - if t.TotalBytes > 0 { - t.Progress = 100 + t.LocalPath = finalPath + t.Format = actual + t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual) + t.DownloadedBytes = n + t.Progress = 100 + if err := a.db.Model(t).Updates(map[string]any{ + "format": t.Format, "local_path": t.LocalPath, + "actual_quality": t.ActualQuality, + "downloaded_bytes": n, "progress": 100, + }).Error; err != nil { + a.invalidateAsset(t) + return + } + if !a.finishAsset(t) { + a.fail(t, fmt.Errorf("shared asset ownership lost before completion")) + return } - a.db.Save(t) - a.markAssetDone(t) - // The credit pays for the successful media download, not the later WebDAV sync. - // Commit before upload so a storage outage cannot turn a successful download into a free retry. if err := a.commitCredit(t); err != nil { a.fail(t, err) return } + t.Status, t.Error = "upload_pending", "" + if err := a.db.Model(t).Updates(map[string]any{ + "status": "upload_pending", "error": "", + }).Error; err != nil { + return + } a.upload(t) } - func (a *App) copyWithProgress(dst *os.File, src io.Reader, t *Task) (int64, error) { var n, last int64 buf := make([]byte, 256*1024) @@ -168,7 +246,12 @@ func (a *App) copyWithProgress(dst *os.File, src io.Reader, t *Task) (int64, err progress = float64(n) * 100 / float64(t.TotalBytes) } t.DownloadedBytes, t.Progress = n, progress - a.db.Model(t).Updates(map[string]any{"downloaded_bytes": n, "progress": progress}) + if err := a.db.Model(t).Updates(map[string]any{ + "downloaded_bytes": n, "progress": progress, + }).Error; err != nil { + return n, err + } + a.markAssetHeartbeat(t.AssetID, t.ID) last = n } } diff --git a/handlers.go b/handlers.go index 325a420..df1dc51 100644 --- a/handlers.go +++ b/handlers.go @@ -6,6 +6,7 @@ import ( "strings" "github.com/gin-gonic/gin" + "gorm.io/gorm" ) func (a *App) search(c *gin.Context) { @@ -60,46 +61,67 @@ func (a *App) addTask(c *gin.Context) { in.Quality = a.cfg.DefaultQuality } user := c.MustGet("user").(*User) + var existing Task - err := a.db.Where("user_id = ? AND source = ? AND r_id = ? AND quality = ?", user.ID, in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error + err := a.db.Where("user_id = ? AND source = ? AND r_id = ? AND quality = ?", + user.ID, in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error if err == nil { - if existing.Status != "error" { + existing.Title, existing.Artist, existing.Album, existing.Pic = in.Title, in.Artist, in.Album, in.Pic + switch existing.Status { + case "pending", "resolving", "downloading", "uploading", "upload_pending": c.JSON(200, existing) return + case "done": + c.JSON(200, existing) + return + case "error": + if err := a.prepareRetry(&existing); err != nil { + if err == errInsufficientCredits { + c.JSON(402, gin.H{"error": err.Error()}) + } else { + c.JSON(409, gin.H{"error": err.Error()}) + } + return + } + a.enqueueTask(existing.ID) + c.JSON(202, existing) + return } - r, reserveErr := a.reserveCredit(user.ID, existing.ID, 1) + } + + t := Task{ + UserID: user.ID, Source: in.Source, RID: in.RID, + Title: in.Title, Artist: in.Artist, Album: in.Album, + Pic: in.Pic, Quality: in.Quality, Status: "pending", + } + var reservation *CreditReservation + err = a.transactionRetry(func(tx *gorm.DB) error { + if err := tx.Create(&t).Error; err != nil { + return err + } + var reserveErr error + reservation, reserveErr = reserveCreditTx(tx, user.ID, t.ID, 1) if reserveErr != nil { - c.JSON(402, gin.H{"error": reserveErr.Error()}) - return + return reserveErr } - a.db.Model(&existing).Updates(map[string]any{"title": in.Title, "artist": in.Artist, "album": in.Album, "pic": in.Pic, "status": "pending", "error": "", "credit_reservation_id": r.ID}) - existing.CreditReservationID = r.ID - a.queue <- existing.ID - c.JSON(202, existing) - return - } - t := Task{UserID: user.ID, Source: in.Source, RID: in.RID, Title: in.Title, Artist: in.Artist, Album: in.Album, Pic: in.Pic, Quality: in.Quality, Status: "pending"} - if err := a.db.Create(&t).Error; err != nil { - if a.db.Where("user_id = ? AND source = ? AND r_id = ? AND quality = ?", user.ID, in.Source, in.RID, in.Quality).Order("id desc").First(&existing).Error == nil { + t.CreditReservationID = reservation.ID + return tx.Model(&t).Update("credit_reservation_id", reservation.ID).Error + }) + if err != nil { + if a.db.Where("user_id = ? AND source = ? AND r_id = ? AND quality = ?", + user.ID, in.Source, in.RID, in.Quality).First(&existing).Error == nil { c.JSON(200, existing) return } + if err == errInsufficientCredits { + c.JSON(402, gin.H{"error": err.Error()}) + return + } c.JSON(500, gin.H{"error": err.Error()}) return } - r, reserveErr := a.reserveCredit(user.ID, t.ID, 1) - if reserveErr != nil { - _ = a.db.Delete(&t).Error - c.JSON(402, gin.H{"error": reserveErr.Error()}) - return - } - t.CreditReservationID = r.ID - if err := a.db.Model(&t).Update("credit_reservation_id", r.ID).Error; err != nil { - _ = a.refundCredit(&t) - c.JSON(500, gin.H{"error": err.Error()}) - return - } - a.queue <- t.ID + + a.enqueueTask(t.ID) c.JSON(202, t) } @@ -119,28 +141,35 @@ func (a *App) retryTask(c *gin.Context) { c.Status(400) return } - var t Task user := c.MustGet("user").(*User) + var t Task if a.db.Where("id = ? AND user_id = ?", id, user.ID).First(&t).Error != nil { c.Status(404) return } - if t.Status != "error" { - c.JSON(409, gin.H{"error": "only failed tasks can be retried"}) + switch t.Status { + case "done": + c.JSON(409, gin.H{"error": "task is already completed"}) + return + case "pending", "resolving", "downloading", "uploading": + c.JSON(409, gin.H{"error": "task is already running"}) + return + case "upload_pending", "error": + if t.Status == "error" { + if err := a.prepareRetry(&t); err != nil { + if err == errInsufficientCredits { + c.JSON(402, gin.H{"error": err.Error()}) + } else { + c.JSON(409, gin.H{"error": err.Error()}) + } + return + } + } + default: + c.JSON(409, gin.H{"error": "task cannot be retried from current state"}) return } - r, reserveErr := a.reserveCredit(user.ID, t.ID, 1) - if reserveErr != nil { - c.JSON(402, gin.H{"error": reserveErr.Error()}) - return - } - if err := a.db.Model(&t).Updates(map[string]any{"status": "pending", "error": "", "progress": 0, "downloaded_bytes": 0, "credit_reservation_id": r.ID}).Error; err != nil { - _ = a.refundCredit(&Task{ID: t.ID, UserID: user.ID, CreditReservationID: r.ID}) - c.JSON(500, gin.H{"error": err.Error()}) - return - } - t.Status, t.Error, t.Progress, t.DownloadedBytes, t.CreditReservationID = "pending", "", 0, 0, r.ID - a.queue <- t.ID + a.enqueueTask(t.ID) c.JSON(202, t) } diff --git a/main.go b/main.go index 79060b2..8afebc2 100644 --- a/main.go +++ b/main.go @@ -13,8 +13,13 @@ import ( func main() { dbPath := getenv("DB_PATH", "/data/downloader.db") _ = os.MkdirAll(filepath.Dir(dbPath), 0755) - db, err := gorm.Open(sqlite.Open(dbPath), &gorm.Config{}) + dsn := dbPath + "?_pragma=journal_mode(WAL)&_pragma=busy_timeout(10000)" + db, err := gorm.Open(sqlite.Open(dsn), &gorm.Config{}) must(err) + sqlDB, err := db.DB() + must(err) + sqlDB.SetMaxOpenConns(8) + sqlDB.SetMaxIdleConns(8) must(migrateDatabase(db)) client := &http.Client{Timeout: 2 * 60 * 1e9} diff --git a/webdav.go b/webdav.go index 5990216..64598b9 100644 --- a/webdav.go +++ b/webdav.go @@ -7,48 +7,63 @@ import ( ) func (a *App) upload(t *Task) { + if t.RemotePath != "" && t.Status == "done" { + return + } storage, root, enabled := a.storageForTask(t) if !enabled || storage == nil || root == "" { - t.Status, t.Progress = "done", 100 - a.db.Save(t) - if err := a.commitCredit(t); err != nil { - a.fail(t, err) - return + if t.Status == "upload_pending" { + if err := a.db.Model(t).Updates(map[string]any{ + "status": "done", "error": "", + }).Error; err != nil { + return + } + t.Status, t.Error = "done", "" } return } - a.db.Model(t).Updates(map[string]any{"status": "uploading"}) - remote := strings.TrimRight(root, "/") + "/" + safe(t.Artist) + "/" + safe(t.Album) + "/" + safe(t.Title) + "." + safe(t.Format) + if !a.claimUpload(t) { + return + } + remote := strings.TrimRight(root, "/") + "/" + safe(t.Artist) + "/" + + safe(t.Album) + "/" + safe(t.Title) + "." + safe(t.Format) if err := storage.MkdirAll(filepath.ToSlash(filepath.Dir(remote))); err != nil { a.failUpload(t, err) return } - if fi, err := os.Stat(t.LocalPath); err == nil { - if same, _ := storage.ExistsSameSize(remote, fi.Size()); same { - t.Status, t.Progress = "done", 100 - a.db.Save(t) - if err := a.commitCredit(t); err != nil { - a.fail(t, err) - return - } - a.cleanupLocalFile(t) + fi, statErr := os.Stat(t.LocalPath) + if statErr != nil { + if a.recoverMissingLocalUpload(t) { return } + a.failUpload(t, statErr) + return + } + if same, _ := storage.ExistsSameSize(remote, fi.Size()); same { + t.RemotePath = remote + t.Status, t.Progress, t.Error = "done", 100, "" + _ = a.db.Model(t).Updates(map[string]any{ + "remote_path": remote, "status": "done", + "progress": 100, "error": "", + }).Error + a.cleanupLocalFile(t) + return } if err := storage.Upload(t.LocalPath, remote); err != nil { a.failUpload(t, err) return } t.RemotePath = remote + t.Status, t.Progress, t.Error = "done", 100, "" + if err := a.db.Model(t).Updates(map[string]any{ + "remote_path": remote, "status": "done", + "progress": 100, "error": "", + }).Error; err != nil { + return + } if a.cfg.WebDAVVersionPolicy == "replace_lower" { a.cleanupLowerQualityVersions(t, storage) } - t.Status, t.Progress = "done", 100 - a.db.Save(t) - if err := a.commitCredit(t); err != nil { - a.fail(t, err) - return - } a.cleanupLocalFile(t) } @@ -56,20 +71,47 @@ func (a *App) cleanupLocalFile(t *Task) { if a.cfg.LocalFilePolicy != "delete_after_upload" || t.LocalPath == "" { return } + if t.AssetID != 0 { + var pending int64 + if err := a.db.Model(&Task{}). + Where("asset_id = ? AND id <> ? AND local_path = ? AND "+ + "(status IN ? OR (status = ? AND (remote_path = '' OR remote_path IS NULL)))", + t.AssetID, t.ID, t.LocalPath, + []string{"pending", "resolving", "downloading", "uploading", "upload_pending"}, + "done").Count(&pending).Error; err != nil { + return + } + if pending > 0 { + return + } + } + if err := os.Remove(t.LocalPath); err != nil && !os.IsNotExist(err) { t.Error = "WebDAV 上传成功,但删除本地文件失败: " + err.Error() - a.db.Save(t) + _ = a.db.Save(t) + return + } + if t.AssetID != 0 { + _ = a.db.Model(&MediaAsset{}). + Where("id = ? AND local_path = ?", t.AssetID, t.LocalPath). + Updates(map[string]any{ + "status": "evicted", "local_path": "", + "size": 0, "owner_task_id": 0, "lease_until": nil, + }).Error } } func (a *App) cleanupLowerQualityVersions(t *Task, storage RemoteStorage) { - levels := map[string]int{"standard": 1, "exhigh": 2, "sq": 3, "lossless": 4, "hires": 5, "jyeffect": 6, "sky": 7, "jymaster": 8, "atmos": 9, "atmos_plus": 10, "master": 11} + levels := map[string]int{"standard": 1, "exhigh": 2, "sq": 3, + "lossless": 4, "hires": 5, "jyeffect": 6, "sky": 7, + "jymaster": 8, "atmos": 9, "atmos_plus": 10, "master": 11} current := qualityRank(levels, t.ActualQuality, t.ResolvedQuality, t.Quality) if current == 0 { return } var old []Task - a.db.Where("user_id = ? AND artist = ? AND title = ? AND album = ? AND status = ? AND id <> ?", t.UserID, t.Artist, t.Title, t.Album, "done", t.ID).Find(&old) + a.db.Where("user_id = ? AND artist = ? AND title = ? AND album = ? "+ + "AND status = ? AND id <> ?", t.UserID, t.Artist, t.Title, t.Album, "done", t.ID).Find(&old) for _, candidate := range old { q := qualityRank(levels, candidate.ActualQuality, candidate.ResolvedQuality, candidate.Quality) if q == 0 || q >= current || candidate.RemotePath == "" || candidate.RemotePath == t.RemotePath { diff --git a/webdav_resync.go b/webdav_resync.go index 43c9726..de1ec6b 100644 --- a/webdav_resync.go +++ b/webdav_resync.go @@ -1,21 +1,125 @@ package main -import "time" +import ( + "os" + "time" + + "gorm.io/gorm" +) + +// recoverMissingLocalUpload re-queues a task when its shared media file has +// disappeared. This can happen after an older version deleted a shared file +// after another user's upload. The asset is invalidated so the next worker +// downloads it again instead of retrying an impossible WebDAV upload forever. +func (a *App) recoverMissingLocalUpload(t *Task) bool { + if t.AssetID == 0 || t.LocalPath == "" { + return false + } + if _, err := os.Stat(t.LocalPath); err == nil || !os.IsNotExist(err) { + return false + } + + var queued bool + err := a.transactionRetry(func(tx *gorm.DB) error { + var asset MediaAsset + if err := tx.First(&asset, t.AssetID).Error; err != nil { + return err + } + if asset.Status == "done" && asset.LocalPath != "" { + if _, statErr := os.Stat(asset.LocalPath); statErr == nil { + if err := tx.Model(&Task{}).Where("id = ?", t.ID).Updates(map[string]any{ + "local_path": asset.LocalPath, "format": asset.Format, + "actual_quality": asset.ActualQuality, + "status": "upload_pending", "error": "共享资源已恢复,正在继续上传", + }).Error; err != nil { + return err + } + queued = true + return nil + } + } + if asset.LocalPath != "" && asset.LocalPath != t.LocalPath { + return nil + } + asset.Status = "error" + asset.OwnerTaskID = 0 + asset.LeaseUntil = time.Time{} + asset.LocalPath = "" + asset.Size = 0 + if err := tx.Save(&asset).Error; err != nil { + return err + } + if err := tx.Model(&Task{}).Where("id = ?", t.ID).Updates(map[string]any{ + "status": "pending", "error": "本地共享文件不存在,正在重新下载", + "progress": 0, "downloaded_bytes": 0, + }).Error; err != nil { + return err + } + queued = true + return nil + }) + if err != nil || !queued { + return false + } + a.enqueueTask(t.ID) + return true +} // resyncPendingUploads repairs tasks that finished locally while WebDAV was // unavailable or the user's WebDAV permission was not yet configured. func (a *App) resyncPendingUploads() { var tasks []Task - if err := a.db.Where("status IN ? AND local_path <> ? AND (remote_path = ? OR remote_path IS NULL)", []string{"done", "upload_pending", "error"}, "", "").Order("id asc").Limit(100).Find(&tasks).Error; err != nil { + if err := a.db.Where("local_path <> ? AND ((status IN ? AND (remote_path = '' OR remote_path IS NULL)) OR status = ?)", + "", []string{"done", "upload_pending"}, "error").Order("id asc").Limit(100).Find(&tasks).Error; err != nil { return } for i := range tasks { t := tasks[i] + if t.Status == "error" { + status, err := a.taskReservationStatus(t.ID, t.CreditReservationID) + if err != nil { + continue + } + if status == "committed" { + if err := a.repairCommittedTask(&t); err != nil { + continue + } + } else { + var asset MediaAsset + if t.AssetID == 0 || a.db.First(&asset, t.AssetID).Error != nil || + asset.Status != "done" || asset.LocalPath == "" { + continue + } + if _, statErr := os.Stat(asset.LocalPath); statErr != nil { + continue + } + t.LocalPath = asset.LocalPath + t.Format = asset.Format + t.ActualQuality = asset.ActualQuality + t.Status = "upload_pending" + t.Error = "共享资源已完成,历史失败任务恢复上传" + if err := a.db.Model(&t).Updates(map[string]any{ + "local_path": t.LocalPath, "format": t.Format, + "actual_quality": t.ActualQuality, + "status": t.Status, "error": t.Error, + }).Error; err != nil { + continue + } + } + } if _, _, ok := a.storageForTask(&t); !ok { continue } - a.upload(&t) - time.Sleep(50 * time.Millisecond) + if a.recoverMissingLocalUpload(&t) { + continue + } + var current Task + if a.db.First(¤t, t.ID).Error != nil { + continue + } + if current.Status == "done" || current.Status == "upload_pending" { + a.enqueueTask(current.ID) + } } } diff --git a/webdav_test.go b/webdav_test.go new file mode 100644 index 0000000..c0bb28f --- /dev/null +++ b/webdav_test.go @@ -0,0 +1,79 @@ +package main + +import ( + "os" + "path/filepath" + "testing" +) + +func TestCleanupLocalFileKeepsSharedAssetForPendingTask(t *testing.T) { + db := openMigrationTestDB(t) + if err := migrateDatabase(db); err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "shared.flac") + if err := os.WriteFile(path, []byte("audio"), 0644); err != nil { + t.Fatal(err) + } + asset := MediaAsset{AssetKey: assetKey("kw", "shared", "lossless"), Source: "kw", RID: "shared", Quality: "lossless", LocalPath: path, Format: "flac", Status: "done", Size: 5} + if err := db.Create(&asset).Error; err != nil { + t.Fatal(err) + } + first := Task{UserID: 1, AssetID: asset.ID, Source: "kw", RID: "shared", Quality: "lossless", Status: "done", LocalPath: path, RemotePath: "Music/shared.flac"} + second := Task{UserID: 2, AssetID: asset.ID, Source: "kw", RID: "shared", Quality: "lossless", Status: "upload_pending", LocalPath: path} + if err := db.Create(&first).Error; err != nil { + t.Fatal(err) + } + if err := db.Create(&second).Error; err != nil { + t.Fatal(err) + } + a := &App{db: db, cfg: Config{LocalFilePolicy: "delete_after_upload"}} + a.cleanupLocalFile(&first) + if _, err := os.Stat(path); err != nil { + t.Fatalf("shared file deleted while needed: %v", err) + } + if err := db.Model(&second).Updates(map[string]any{"status": "done", "remote_path": "Music/shared.flac"}).Error; err != nil { + t.Fatal(err) + } + a.cleanupLocalFile(&first) + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("shared file still exists: %v", err) + } + var gotAsset MediaAsset + if err := db.First(&gotAsset, asset.ID).Error; err != nil { + t.Fatal(err) + } + if gotAsset.Status != "evicted" || gotAsset.LocalPath != "" { + t.Fatalf("asset not evicted: %+v", gotAsset) + } +} + +func TestRecoverMissingLocalUploadRequeuesTask(t *testing.T) { + db := openMigrationTestDB(t) + if err := migrateDatabase(db); err != nil { + t.Fatal(err) + } + path := filepath.Join(t.TempDir(), "missing.flac") + asset := MediaAsset{AssetKey: assetKey("kw", "missing", "lossless"), Source: "kw", RID: "missing", Quality: "lossless", LocalPath: path, Format: "flac", Status: "done", Size: 10} + if err := db.Create(&asset).Error; err != nil { + t.Fatal(err) + } + task := Task{UserID: 1, AssetID: asset.ID, Source: "kw", RID: "missing", Quality: "lossless", Status: "upload_pending", LocalPath: path} + if err := db.Create(&task).Error; err != nil { + t.Fatal(err) + } + a := &App{db: db, queue: make(chan uint, 1)} + if !a.recoverMissingLocalUpload(&task) { + t.Fatal("missing local file was not requeued") + } + var got Task + if err := db.First(&got, task.ID).Error; err != nil { + t.Fatal(err) + } + if got.Status != "pending" { + t.Fatalf("task status = %q, want pending", got.Status) + } + if <-a.queue != task.ID { + t.Fatal("queued task id mismatch") + } +}