feat: harden multi-user download concurrency

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 隔离。

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(&current, 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(&current).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) }

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(&current, 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(&current).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<<attempt)) * time.Millisecond)
+	}
+	return err
+}

concurrency_test.go

@@ -0,0 +1,223 @@
+package main
+
+import (
+	"os"
+	"path/filepath"
+	"strconv"
+	"sync"
+	"testing"
+
+	"github.com/glebarez/sqlite"
+	"gorm.io/gorm"
+	"time"
+)
+
+func TestClaimTaskIsAtomic(t *testing.T) {
+	a := newCreditTestApp(t)
+	task := Task{UserID: 1, Source: "kw", RID: "atomic", Quality: "lossless", Status: "pending"}
+	if err := a.db.Create(&task).Error; err != nil {
+		t.Fatal(err)
+	}
+	if _, err := a.claimTask(task.ID); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := a.claimTask(task.ID); err != errTaskRunning {
+		t.Fatalf("second claim error = %v, want task running", err)
+	}
+}
+
+func TestRetryCommittedReservationDoesNotChargeAgain(t *testing.T) {
+	a := newCreditTestApp(t)
+	user := User{Username: "retry-committed", Role: RoleUser, Status: UserActive}
+	if err := a.db.Create(&user).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := a.db.Create(&CreditAccount{UserID: user.ID, Balance: 1}).Error; err != nil {
+		t.Fatal(err)
+	}
+	task := Task{UserID: user.ID, Source: "kw", RID: "committed", Quality: "lossless", Status: "pending"}
+	if err := a.db.Create(&task).Error; err != nil {
+		t.Fatal(err)
+	}
+	r, err := a.reserveCredit(user.ID, task.ID, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	task.CreditReservationID = r.ID
+	if err := a.db.Model(&task).Update("credit_reservation_id", r.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := a.commitCredit(&task); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.db.Model(&task).Updates(map[string]any{"status": "error", "error": "legacy"}).Error; err != nil {
+		t.Fatal(err)
+	}
+	var current Task
+	if err := a.db.First(&current, task.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := a.prepareRetry(&current); err != nil {
+		t.Fatal(err)
+	}
+	var account CreditAccount
+	if err := a.db.Where("user_id = ?", user.ID).First(&account).Error; err != nil {
+		t.Fatal(err)
+	}
+	if account.Balance != 0 || account.Reserved != 0 {
+		t.Fatalf("credit changed on committed retry: %+v", account)
+	}
+	if current.Status != "upload_pending" {
+		t.Fatalf("status = %q, want upload_pending", current.Status)
+	}
+}
+
+func TestRetryRefundedReservationCreatesNewReservation(t *testing.T) {
+	a := newCreditTestApp(t)
+	user := User{Username: "retry-refunded", Role: RoleUser, Status: UserActive}
+	if err := a.db.Create(&user).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := a.db.Create(&CreditAccount{UserID: user.ID, Balance: 1}).Error; err != nil {
+		t.Fatal(err)
+	}
+	task := Task{UserID: user.ID, Source: "kw", RID: "refunded", Quality: "lossless", Status: "error"}
+	if err := a.db.Create(&task).Error; err != nil {
+		t.Fatal(err)
+	}
+	r1, err := a.reserveCredit(user.ID, task.ID, 1)
+	if err != nil {
+		t.Fatal(err)
+	}
+	task.CreditReservationID = r1.ID
+	if err := a.refundCredit(&task); err != nil {
+		t.Fatal(err)
+	}
+	if err := a.prepareRetry(&task); err != nil {
+		t.Fatal(err)
+	}
+	if task.CreditReservationID == r1.ID {
+		t.Fatal("retry reused refunded reservation")
+	}
+	var account CreditAccount
+	if err := a.db.Where("user_id = ?", user.ID).First(&account).Error; err != nil {
+		t.Fatal(err)
+	}
+	if account.Balance != 1 || account.Reserved != 1 {
+		t.Fatalf("unexpected account: %+v", account)
+	}
+}
+
+func TestStaleAssetLeaseStealsSafely(t *testing.T) {
+	a := newCreditTestApp(t)
+	root := t.TempDir()
+	a.cfg.DownloadDir = root
+	first := Task{UserID: 1, Source: "kw", RID: "lease", Quality: "lossless", Format: "flac", Status: "downloading"}
+	second := Task{UserID: 2, Source: "kw", RID: "lease", Quality: "lossless", Format: "flac", Status: "pending"}
+	if err := a.db.Create(&first).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := a.db.Create(&second).Error; err != nil {
+		t.Fatal(err)
+	}
+	asset := MediaAsset{
+		AssetKey: assetKey("kw", "lease", "lossless"),
+		Source:   "kw", RID: "lease", Quality: "lossless",
+		LocalPath: filepath.Join(root, "old.flac"), Format: "flac",
+		Status: "downloading", OwnerTaskID: first.ID,
+		LeaseUntil: time.Now().Add(-time.Minute), Generation: 1,
+	}
+	if err := a.db.Create(&asset).Error; err != nil {
+		t.Fatal(err)
+	}
+	first.AssetID = asset.ID
+	if err := a.db.Model(&first).Update("asset_id", asset.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	got, reused, err := a.acquireAsset(&second)
+	if err != nil || reused {
+		t.Fatalf("acquire = asset=%+v reused=%v err=%v", got, reused, err)
+	}
+	if got.Generation != 2 {
+		t.Fatalf("generation = %d, want 2", got.Generation)
+	}
+	if got.LocalPath == asset.LocalPath {
+		t.Fatal("stolen generation reused old local path")
+	}
+	var old Task
+	if err := a.db.First(&old, first.ID).Error; err != nil {
+		t.Fatal(err)
+	}
+	if old.Status != "pending" {
+		t.Fatalf("stale owner status = %q, want pending", old.Status)
+	}
+}
+
+func TestAssetCachePathIsStablePerGeneration(t *testing.T) {
+	root := t.TempDir()
+	p1 := assetCachePath(root, "kw-rid-lossless", "flac", 1)
+	p2 := assetCachePath(root, "kw-rid-lossless", "flac", 2)
+	if p1 == p2 || filepath.Dir(p1) != filepath.Dir(p2) {
+		t.Fatalf("unexpected cache paths: %q %q", p1, p2)
+	}
+	if err := os.MkdirAll(filepath.Dir(p1), 0755); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestConcurrentCreditReservationsAreAtomic(t *testing.T) {
+	dbPath := filepath.Join(t.TempDir(), "credits.db")
+	db, err := gorm.Open(sqlite.Open(dbPath+"?_pragma=journal_mode(WAL)&_pragma=busy_timeout(10000)"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := db.AutoMigrate(&Task{}, &MediaAsset{}, &User{}, &CreditAccount{}, &CreditTransaction{}, &CreditReservation{}); err != nil {
+		t.Fatal(err)
+	}
+	app := &App{db: db}
+	user := User{Username: "concurrent-credit", Role: RoleUser, Status: UserActive}
+	if err := db.Create(&user).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err := db.Create(&CreditAccount{UserID: user.ID, Balance: 5}).Error; err != nil {
+		t.Fatal(err)
+	}
+	for i := 0; i < 20; i++ {
+		if err := db.Create(&Task{UserID: user.ID, Source: "kw", RID: "c-" + strconv.Itoa(i), Quality: "lossless", Status: "pending"}).Error; err != nil {
+			t.Fatal(err)
+		}
+	}
+	var wg sync.WaitGroup
+	var mu sync.Mutex
+	success, unexpected := 0, 0
+	for i := 0; i < 20; i++ {
+		var task Task
+		if err := db.Where("user_id = ? AND r_id = ?", user.ID, "c-"+strconv.Itoa(i)).First(&task).Error; err != nil {
+			t.Fatal(err)
+		}
+		wg.Add(1)
+		go func(taskID uint) {
+			defer wg.Done()
+			if _, err := app.reserveCredit(user.ID, taskID, 1); err == nil {
+				mu.Lock()
+				success++
+				mu.Unlock()
+			} else if err != errInsufficientCredits {
+				mu.Lock()
+				unexpected++
+				mu.Unlock()
+			}
+		}(task.ID)
+	}
+	wg.Wait()
+	if success != 5 || unexpected != 0 {
+		t.Fatalf("reservations: success=%d unexpected=%d", success, unexpected)
+	}
+	var account CreditAccount
+	if err := db.Where("user_id = ?", user.ID).First(&account).Error; err != nil {
+		t.Fatal(err)
+	}
+	if account.Balance != 5 || account.Reserved != 5 {
+		t.Fatalf("account after concurrent reservations: %+v", account)
+	}
+}

credits.go

@@ -4,6 +4,7 @@ import (
 	"errors"
 	"fmt"
 	"strconv"
+	"strings"
 	"time"
 
 	"gorm.io/gorm"
@@ -16,50 +17,76 @@ var (
 
 // reserveCredit places one credit on hold for a task. The hold is released on
 // failure or converted into a debit after a successful new download.
-func (a *App) reserveCredit(userID, taskID uint, amount int64) (*CreditReservation, error) {
+func reserveCreditTx(tx *gorm.DB, userID, taskID uint, amount int64) (*CreditReservation, error) {
 	if amount <= 0 {
 		return nil, fmt.Errorf("invalid credit amount")
 	}
-	var reservation CreditReservation
-	err := a.db.Transaction(func(tx *gorm.DB) error {
 	var account CreditAccount
 	if err := tx.Where("user_id = ?", userID).First(&account).Error; err != nil {
 		if !errors.Is(err, gorm.ErrRecordNotFound) {
-				return err
+			return nil, err
 		}
 		account = CreditAccount{UserID: userID}
 		if err := tx.Create(&account).Error; err != nil {
-				return err
+			if err := tx.Where("user_id = ?", userID).First(&account).Error; err != nil {
+				return nil, err
+			}
 		}
 	}
 	var existing CreditReservation
-		if err := tx.Where("task_id = ? AND status = ?", taskID, "reserved").Order("id desc").First(&existing).Error; err == nil {
-			reservation = existing
-			return nil
+	if err := tx.Where("task_id = ? AND status = ?", taskID, "reserved").
+		Order("id desc").First(&existing).Error; err == nil {
+		return &existing, nil
 	} else if !errors.Is(err, gorm.ErrRecordNotFound) {
-			return err
+		return nil, err
 	}
 	if account.Balance-account.Reserved < amount {
-			return errInsufficientCredits
+		return nil, errInsufficientCredits
 	}
 	account.Reserved += amount
 	if err := tx.Save(&account).Error; err != nil {
-			return err
+		return nil, err
 	}
-		reservation = CreditReservation{UserID: userID, TaskID: taskID, Amount: amount, Status: "reserved"}
-		if err := tx.Create(&reservation).Error; err != nil {
-			return err
+	reservation := &CreditReservation{
+		UserID: userID, TaskID: taskID, Amount: amount, Status: "reserved",
 	}
-		return tx.Create(&CreditTransaction{UserID: userID, Type: "reserve", Amount: 0, BalanceAfter: account.Balance, ReferenceType: "task", ReferenceID: strconv.FormatUint(uint64(taskID), 10), Reason: "download reservation"}).Error
+	if err := tx.Create(reservation).Error; err != nil {
+		return nil, err
+	}
+	if err := tx.Create(&CreditTransaction{
+		UserID: userID, Type: "reserve", Amount: 0,
+		BalanceAfter: account.Balance, ReferenceType: "task",
+		ReferenceID: strconv.FormatUint(uint64(taskID), 10),
+		Reason:      "download reservation",
+	}).Error; err != nil {
+		return nil, err
+	}
+	return reservation, nil
+}
+
+func isSQLiteBusy(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "sqlite_busy") || strings.Contains(msg, "database is locked")
+}
+
+func (a *App) reserveCredit(userID, taskID uint, amount int64) (*CreditReservation, error) {
+	var reservation *CreditReservation
+	err := a.transactionRetry(func(tx *gorm.DB) error {
+		var txErr error
+		reservation, txErr = reserveCreditTx(tx, userID, taskID, amount)
+		return txErr
 	})
-	return &reservation, err
+	return reservation, err
 }
 
 func (a *App) commitCredit(task *Task) error {
 	if task.CreditReservationID == 0 {
 		return nil
 	}
-	return a.db.Transaction(func(tx *gorm.DB) error {
+	return a.transactionRetry(func(tx *gorm.DB) error {
 		var r CreditReservation
 		if err := tx.First(&r, task.CreditReservationID).Error; err != nil {
 			return err
@@ -97,7 +124,7 @@ func (a *App) refundCredit(task *Task) error {
 	if task.CreditReservationID == 0 {
 		return nil
 	}
-	return a.db.Transaction(func(tx *gorm.DB) error {
+	return a.transactionRetry(func(tx *gorm.DB) error {
 		var r CreditReservation
 		if err := tx.First(&r, task.CreditReservationID).Error; err != nil {
 			return err

downloader.go

@@ -7,6 +7,7 @@ import (
 	"os"
 	"path/filepath"
 	"strings"
+	"time"
 )
 
 func (a *App) startWorkers() {
@@ -15,21 +16,36 @@ func (a *App) startWorkers() {
 		go func() {
 			defer a.wg.Done()
 			for id := range a.queue {
-				a.download(id)
+				a.processTask(id)
 			}
 		}()
 	}
 }
 
-func (a *App) download(id uint) {
+func (a *App) processTask(id uint) {
 	var t Task
-	if a.db.First(&t, id).Error != nil || t.Status != "pending" {
+	if a.db.First(&t, id).Error != nil {
 		return
 	}
-	a.db.Model(&t).Updates(map[string]any{"status": "resolving", "error": "", "progress": 0})
+	if t.Status == "upload_pending" {
+		a.upload(&t)
+		return
+	}
+	if t.Status != "pending" {
+		return
+	}
+	a.download(id)
+}
+
+func (a *App) download(id uint) {
+	t, err := a.claimTask(id)
+	if err != nil {
+		return
+	}
+
 	info, err := a.am.Resolve(t.Source, t.RID, t.Quality)
 	if err != nil {
-		a.fail(&t, err)
+		a.fail(t, err)
 		return
 	}
 
@@ -41,17 +57,30 @@ func (a *App) download(id uint) {
 		ext = "mp3"
 	}
 	t.Format, t.ResolvedQuality, t.ActualQuality = ext, info.Level, ""
-	asset, reused, err := a.acquireAsset(&t)
+	if err := a.db.Model(t).Updates(map[string]any{
+		"format": t.Format, "resolved_quality": t.ResolvedQuality,
+	}).Error; err != nil {
+		a.fail(t, err)
+		return
+	}
+
+	asset, reused, err := a.acquireAsset(t)
 	if err != nil {
-		a.fail(&t, err)
+		a.fail(t, err)
 		return
 	}
 	if reused {
-		a.finishExisting(&t, asset.Size)
+		a.finishExisting(t, asset.Size)
 		return
 	}
-	a.db.Model(&t).Updates(map[string]any{"format": t.Format, "local_path": t.LocalPath})
-	a.downloadFile(&t, info)
+	if err := a.db.Model(t).Updates(map[string]any{
+		"local_path": t.LocalPath, "status": "downloading",
+	}).Error; err != nil {
+		a.invalidateAsset(t)
+		a.fail(t, err)
+		return
+	}
+	a.downloadFile(t, info)
 }
 
 func (a *App) finishExisting(t *Task, size int64) {
@@ -70,8 +99,9 @@ func (a *App) finishExisting(t *Task, size int64) {
 		t.ActualQuality = actualQuality(t.Quality, t.ResolvedQuality, actual)
 	}
 	t.TotalBytes, t.DownloadedBytes, t.Progress, t.Status = size, size, 100, "done"
-	a.db.Save(t)
-	// Existing local media is reused and therefore does not consume a credit.
+	if err := a.db.Save(t).Error; err != nil {
+		return
+	}
 	if err := a.refundCredit(t); err != nil {
 		a.fail(t, err)
 		return
@@ -80,7 +110,6 @@ func (a *App) finishExisting(t *Task, size int64) {
 }
 
 func (a *App) downloadFile(t *Task, info playInfo) {
-	a.db.Model(t).Updates(map[string]any{"status": "downloading", "format": t.Format, "local_path": t.LocalPath})
 	req, err := http.NewRequest("GET", info.URL, nil)
 	if err != nil {
 		a.fail(t, err)
@@ -99,27 +128,62 @@ func (a *App) downloadFile(t *Task, info playInfo) {
 		a.fail(t, fmt.Errorf("media http %d", resp.StatusCode))
 		return
 	}
-	t.Status, t.TotalBytes, t.DownloadedBytes, t.Progress = "downloading", resp.ContentLength, 0, 0
-	a.db.Save(t)
 
-	tmp := t.LocalPath + ".part"
+	t.Status = "downloading"
+	t.TotalBytes = resp.ContentLength
+	t.DownloadedBytes = 0
+	t.Progress = 0
+	_ = a.db.Model(t).Updates(map[string]any{
+		"status": t.Status, "total_bytes": t.TotalBytes,
+		"downloaded_bytes": 0, "progress": 0,
+	}).Error
+
+	if err := os.MkdirAll(filepath.Dir(t.LocalPath), 0755); err != nil {
+		a.fail(t, err)
+		return
+	}
+	tmp := fmt.Sprintf("%s.task-%d.part", t.LocalPath, t.ID)
+	_ = os.Remove(tmp)
 	f, err := os.Create(tmp)
 	if err != nil {
 		a.fail(t, err)
 		return
 	}
-	n, err := a.copyWithProgress(f, resp.Body, t)
+
+	stop := make(chan struct{})
+	ticker := time.NewTicker(30 * time.Second)
+	go func() {
+		for {
+			select {
+			case <-ticker.C:
+				a.markAssetHeartbeat(t.AssetID, t.ID)
+			case <-stop:
+				return
+			}
+		}
+	}()
+	n, copyErr := a.copyWithProgress(f, resp.Body, t)
 	closeErr := f.Close()
-	if err != nil {
-		a.fail(t, err)
+	ticker.Stop()
+	close(stop)
+
+	if copyErr != nil {
+		a.fail(t, copyErr)
+		_ = os.Remove(tmp)
 		return
 	}
 	if closeErr != nil {
 		a.fail(t, closeErr)
+		_ = os.Remove(tmp)
 		return
 	}
 	if t.TotalBytes > 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.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
 			}
 		}

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()})
 				}
-		r, reserveErr := a.reserveCredit(user.ID, existing.ID, 1)
-		if reserveErr != nil {
-			c.JSON(402, gin.H{"error": reserveErr.Error()})
 				return
 			}
-		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
+			a.enqueueTask(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 := 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 {
+			return reserveErr
+		}
+		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
 			}
-	r, reserveErr := a.reserveCredit(user.ID, t.ID, 1)
-	if reserveErr != nil {
-		c.JSON(402, gin.H{"error": reserveErr.Error()})
+		}
+	default:
+		c.JSON(409, gin.H{"error": "task cannot be retried from current state"})
 		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)
 }
 

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}

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)
+		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)
+	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 {

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(&current, t.ID).Error != nil {
+			continue
+		}
+		if current.Status == "done" || current.Status == "upload_pending" {
+			a.enqueueTask(current.ID)
+		}
 	}
 }
 

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")
+	}
+}