From e113c6c2cfa655a8e3abd7499cd7bbfcf5d13ee2 Mon Sep 17 00:00:00 2001 From: saltbo Date: Fri, 5 Jun 2026 22:19:14 -0400 Subject: [PATCH] fix(downloader): clean recovered torrent runtime --- downloader/internal/engine/aria2.go | 54 ++++++++++++++---- downloader/internal/engine/engine.go | 15 ----- downloader/internal/engine/engine_test.go | 39 +++++++++++++ downloader/internal/engine/http.go | 2 +- downloader/internal/engine/qbittorrent.go | 21 ++++--- downloader/internal/worker/worker.go | 10 +++- downloader/internal/worker/worker_test.go | 56 +++++++++++++++++++ .../routes/download-tasks.integration.test.ts | 22 +++++++- server/services/downloads/core.ts | 44 ++++++++++++++- 9 files changed, 220 insertions(+), 43 deletions(-) diff --git a/downloader/internal/engine/aria2.go b/downloader/internal/engine/aria2.go index 8faf30c2..8991266b 100644 --- a/downloader/internal/engine/aria2.go +++ b/downloader/internal/engine/aria2.go @@ -163,16 +163,32 @@ func (a Aria2) Recover(ctx context.Context, task client.DownloadTask) (Result, b if findErr != nil { return Result{}, false, findErr } - if ok && string(status.Status) == string(arigo.StatusCompleted) { + if ok && isAria2DownloadComplete(status) { files, err := a.getAria2Files(ctx, &aria, status.GID) if err != nil { return Result{}, false, err } - result, err := resultFromAria2Files(task, aria2StatusTaskDir(status, filepath.Join(a.Dir, task.ID)), status.BitTorrent.Info.Name, files) + taskDir := aria2StatusTaskDir(status, filepath.Join(a.Dir, task.ID)) + result, err := resultFromAria2Files(task, taskDir, status.BitTorrent.Info.Name, files) + if err == nil && a.RetainSeed && task.SourceType != "http" { + result.Seed = a.seedFromStatus(status, taskDir) + } return result, err == nil, err } + if ok { + return Result{}, false, fmt.Errorf( + "aria2 task %s is not a completed upload resume candidate: status=%s completed=%d total=%d", + status.GID, + status.Status, + int64(status.CompletedLength), + int64(status.TotalLength), + ) + } } - return recoverFromTaskDir(task, a.Dir) + if err != nil { + return Result{}, false, err + } + return Result{}, false, nil } func (a Aria2) Download(ctx context.Context, task client.DownloadTask, progress Progress) (Result, error) { @@ -220,6 +236,9 @@ func (a Aria2) Download(ctx context.Context, task client.DownloadTask, progress } gid, err := addAria2Task(ctx, aria, task, options) if err != nil { + if !isAria2InfoHashAlreadyRegistered(err) { + return Result{}, fmt.Errorf("add aria2 task: %w", err) + } status, ok, findErr := a.findTask(ctx, &aria, task) if findErr != nil { return Result{}, fmt.Errorf("find aria2 task after add failed: %w", findErr) @@ -303,14 +322,7 @@ func (a Aria2) waitResult(ctx context.Context, aria **arigo.Client, task client. return Result{}, fmt.Errorf("build result from aria2 files: %w", err) } if a.RetainSeed && task.SourceType != "http" { - result.Seed = &Seed{ - Engine: "aria2", - ID: resultGID, - InfoHash: strings.ToLower(status.InfoHash), - Path: taskDir, - Snapshot: a.seedSnapshot(resultGID), - Cleanup: a.cleanupSeed(resultGID, taskDir), - } + result.Seed = a.seedFromStatus(status, taskDir) return result, nil } _ = (*aria).ForceRemove(resultGID) @@ -322,6 +334,26 @@ func (a Aria2) waitResult(ctx context.Context, aria **arigo.Client, task client. return result, nil } +func (a Aria2) seedFromStatus(status arigo.Status, taskDir string) *Seed { + return &Seed{ + Engine: "aria2", + ID: status.GID, + InfoHash: strings.ToLower(status.InfoHash), + Path: taskDir, + Snapshot: a.seedSnapshot(status.GID), + Cleanup: a.cleanupSeed(status.GID, taskDir), + } +} + +func isAria2DownloadComplete(status arigo.Status) bool { + total := int64(status.TotalLength) + completed := int64(status.CompletedLength) + if total > 0 && completed >= total && hasAria2LocalFile(status.Files) { + return true + } + return string(status.Status) == string(arigo.StatusCompleted) +} + func (a Aria2) findSeed(ctx context.Context, aria **arigo.Client, ref SeedRef) (arigo.Status, bool, error) { if ref.ID != "" { status, err := tellAria2Status(*aria, ref.ID) diff --git a/downloader/internal/engine/engine.go b/downloader/internal/engine/engine.go index c873062f..d202db8a 100644 --- a/downloader/internal/engine/engine.go +++ b/downloader/internal/engine/engine.go @@ -125,21 +125,6 @@ func resultFromPath(task client.DownloadTask, path string, fallbackName string) return Result{Path: path, Name: outputName(task, fallbackName), Size: size, IsDir: true}, nil } -func recoverFromTaskDir(task client.DownloadTask, dir string) (Result, bool, error) { - taskDir := filepath.Join(dir, task.ID) - if _, err := os.Stat(taskDir); err != nil { - if os.IsNotExist(err) { - return Result{}, false, nil - } - return Result{}, false, err - } - result, err := resultFromPath(task, taskDir, requestedOutputName(task)) - if err != nil { - return Result{}, false, err - } - return result, true, nil -} - type downloadedFile struct { path string relativePath string diff --git a/downloader/internal/engine/engine_test.go b/downloader/internal/engine/engine_test.go index f4c04d4b..5c3efd08 100644 --- a/downloader/internal/engine/engine_test.go +++ b/downloader/internal/engine/engine_test.go @@ -116,6 +116,30 @@ func TestHTTPDownloadResumesExistingFile(t *testing.T) { } } +func TestHTTPRecoverDoesNotTrustLocalTaskDirectory(t *testing.T) { + dir := t.TempDir() + taskDir := filepath.Join(dir, "task-1") + if err := os.MkdirAll(taskDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(taskDir, "payload.bin"), []byte("payload"), 0o644); err != nil { + t.Fatal(err) + } + + result, recovered, err := (HTTP{Dir: dir}).Recover(context.Background(), client.DownloadTask{ + ID: "task-1", + SourceType: "http", + SourceURI: "https://example.com/payload.bin", + }) + + if err != nil { + t.Fatal(err) + } + if recovered { + t.Fatalf("expected local directory not to be treated as a completed runtime result, got %#v", result) + } +} + func TestAria2StatusKeysRequestBittorrentPayload(t *testing.T) { keys := strings.Join(aria2StatusKeys, ",") if !strings.Contains(keys, "bittorrent") { @@ -259,6 +283,21 @@ func TestAria2StatusMatchesTaskByInfoHash(t *testing.T) { } } +func TestIsAria2DownloadCompleteTreatsActiveFullTorrentAsComplete(t *testing.T) { + status := arigo.Status{ + Status: arigo.StatusActive, + TotalLength: 100, + CompletedLength: 100, + Files: []arigo.File{ + {Path: "/tmp/zpan/task-1/movie.mkv", Length: 100, CompletedLength: 100}, + }, + } + + if !isAria2DownloadComplete(status) { + t.Fatal("expected active full torrent to be recoverable as a completed download") + } +} + func TestIsAria2InfoHashAlreadyRegistered(t *testing.T) { err := errors.New("InfoHash 0546769f209ec059284b47f68659791a6f75ca8e is already registered.") if !isAria2InfoHashAlreadyRegistered(err) { diff --git a/downloader/internal/engine/http.go b/downloader/internal/engine/http.go index f6ad9ab7..6b6e909b 100644 --- a/downloader/internal/engine/http.go +++ b/downloader/internal/engine/http.go @@ -41,7 +41,7 @@ func (h HTTP) Check(ctx context.Context) error { } func (h HTTP) Recover(ctx context.Context, task client.DownloadTask) (Result, bool, error) { - return recoverFromTaskDir(task, h.Dir) + return Result{}, false, nil } func (h HTTP) Download(ctx context.Context, task client.DownloadTask, progress Progress) (Result, error) { diff --git a/downloader/internal/engine/qbittorrent.go b/downloader/internal/engine/qbittorrent.go index 3e937ab2..97d2f25d 100644 --- a/downloader/internal/engine/qbittorrent.go +++ b/downloader/internal/engine/qbittorrent.go @@ -131,7 +131,17 @@ func (q QBittorrent) Recover(ctx context.Context, task client.DownloadTask) (Res result, err := resultFromQBittorrentFiles(ctx, qbt, task, filepath.Join(q.Dir, task.ID), torrent) return result, err == nil, err } - return recoverFromTaskDir(task, q.Dir) + if ok { + return Result{}, false, fmt.Errorf( + "qbittorrent task %s is not a completed upload resume candidate: state=%s progress=%f amount_left=%d total=%d", + torrent.Hash, + torrent.State, + torrent.Progress, + torrent.AmountLeft, + torrent.TotalSize, + ) + } + return Result{}, false, nil } func (q QBittorrent) login(ctx context.Context) (*qbittorrent.Client, error) { @@ -177,14 +187,7 @@ func (q QBittorrent) Download(ctx context.Context, task client.DownloadTask, pro options := qbittorrentAddOptions(task, taskDir, tag) if _, err := qbt.AddTorrentFromUrlCtx(ctx, task.SourceURI, options); err != nil { - torrent, ok, findErr := q.findTask(ctx, qbt, task) - if findErr != nil { - return Result{}, findErr - } - if !ok { - return Result{}, err - } - _ = qbt.StartCtx(ctx, []string{torrent.Hash}) + return Result{}, err } torrent, err = waitQBittorrent(ctx, qbt, tag, progress) diff --git a/downloader/internal/worker/worker.go b/downloader/internal/worker/worker.go index 89d72bf7..fae0a1ea 100644 --- a/downloader/internal/worker/worker.go +++ b/downloader/internal/worker/worker.go @@ -187,7 +187,8 @@ func (w *Worker) process(ctx context.Context, task client.DownloadTask) { log := w.taskLogger(task) log.Info("task started", "source_uri", task.SourceURI, "target_folder", task.TargetFolder) currentDetail := task.Detail - if resumeStage(task) == taskResumeUpload { + stage := resumeStage(task) + if stage == taskResumeUpload { result, recovered, err := w.engine.Recover(ctx, task) if err != nil { msg := taskErrorMessage(err) @@ -202,7 +203,12 @@ func (w *Worker) process(ctx context.Context, task client.DownloadTask) { w.uploadAndComplete(ctx, log, task, result, currentDetail) return } - log.Warn("task has no recoverable completed download result; restarting download", "status", task.Status) + msg := "completed download is not recoverable from downloader runtime" + log.Error("failed to recover completed download result", "error", msg) + if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "failed", ErrorMessage: &msg}); updateErr != nil { + log.Error("failed to mark task failed", "error", updateErr) + } + return } if _, err := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "downloading"}); err != nil { diff --git a/downloader/internal/worker/worker_test.go b/downloader/internal/worker/worker_test.go index f9c98122..57105af0 100644 --- a/downloader/internal/worker/worker_test.go +++ b/downloader/internal/worker/worker_test.go @@ -256,6 +256,58 @@ func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) { } } +func TestUploadResumeRecoveryErrorFailsWithoutRedownloading(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{recoverErr: errors.New("runtime state is inconsistent")} + w := NewWithAPI(config.Config{}, api) + w.engine = eng + + total := int64(100) + w.process(context.Background(), client.DownloadTask{ + ID: "task-1", + Status: "assigned", + SourceType: "magnet", + SourceURI: "magnet:?xt=urn:btih:abc123", + TotalBytes: &total, + Detail: &client.DownloadTaskDetail{Phase: "uploading"}, + UploadToken: "upload-token", + }) + + if eng.downloadCalls != 0 { + t.Fatalf("expected upload resume recovery failure not to restart download, got %d download calls", eng.downloadCalls) + } + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "runtime state is inconsistent") { + t.Fatalf("expected recovery error to be reported, got %#v", failed.ErrorMessage) + } +} + +func TestUploadResumeMissingRuntimeFailsWithoutRedownloading(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{recovered: false} + w := NewWithAPI(config.Config{}, api) + w.engine = eng + + total := int64(100) + w.process(context.Background(), client.DownloadTask{ + ID: "task-1", + Status: "assigned", + SourceType: "magnet", + SourceURI: "magnet:?xt=urn:btih:abc123", + TotalBytes: &total, + Detail: &client.DownloadTaskDetail{Phase: "uploading"}, + UploadToken: "upload-token", + }) + + if eng.downloadCalls != 0 { + t.Fatalf("expected missing runtime during upload resume not to restart download, got %d download calls", eng.downloadCalls) + } + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "not recoverable") { + t.Fatalf("expected missing runtime to be reported, got %#v", failed.ErrorMessage) + } +} + func TestDownloadShutdownMarksTaskInterrupted(t *testing.T) { api := &recordingAPI{} eng := &recordingEngine{downloadErr: context.Canceled} @@ -680,6 +732,7 @@ type recordingEngine struct { downloadResult engine.Result downloadErr error recoverResult engine.Result + recoverErr error recovered bool restoreSeed *engine.Seed downloadCalls int @@ -701,6 +754,9 @@ func (e *recordingEngine) Check(context.Context) error { func (e *recordingEngine) Recover(context.Context, client.DownloadTask) (engine.Result, bool, error) { e.recoverCalls++ + if e.recoverErr != nil { + return engine.Result{}, false, e.recoverErr + } return e.recoverResult, e.recovered, nil } diff --git a/server/routes/download-tasks.integration.test.ts b/server/routes/download-tasks.integration.test.ts index f195a795..7eaba2ac 100644 --- a/server/routes/download-tasks.integration.test.ts +++ b/server/routes/download-tasks.integration.test.ts @@ -683,7 +683,13 @@ describe('Download tasks API integration', () => { body: JSON.stringify({ status: 'downloading', downloadedBytes: 2048, downloadBps: 256 }), }) expect(resumedProgressRes.status).toBe(200) - await expect(resumedProgressRes.json()).resolves.toMatchObject({ status: 'downloading', downloadedBytes: 2048 }) + const resumedProgress = (await resumedProgressRes.json()) as { + status: string + downloadedBytes: number + detail: { message?: string } | null + } + expect(resumedProgress).toMatchObject({ status: 'downloading', downloadedBytes: 2048 }) + expect(resumedProgress.detail?.message).toBeUndefined() const pauseRes = await app.request(`/api/download-tasks/${createdTask.id}/actions`, { method: 'POST', @@ -748,7 +754,7 @@ describe('Download tasks API integration', () => { totalBytes, storageUploadedBytes: 1024, errorMessage: 'confirm object failed', - detail: { engine: 'aria2', phase: 'uploading', infoHash: 'abc123' }, + detail: { engine: 'aria2', phase: 'uploading', infoHash: 'abc123', message: 'upload token rejected' }, }), }) expect(failedUploadRes.status).toBe(200) @@ -766,7 +772,16 @@ describe('Download tasks API integration', () => { body: JSON.stringify({ action: 'retry' }), }) expect(retryRes.status).toBe(200) - await expect(retryRes.json()).resolves.toMatchObject({ + const retriedTask = (await retryRes.json()) as { + status: string + assignedDownloaderId: string + downloadedBytes: number + totalBytes: number + storageUploadedBytes: number + detail: { phase: string; message?: string } | null + errorMessage: string | null + } + expect(retriedTask).toMatchObject({ status: 'assigned', assignedDownloaderId: createdDownloader.downloader.id, downloadedBytes: totalBytes, @@ -775,6 +790,7 @@ describe('Download tasks API integration', () => { detail: { phase: 'uploading' }, errorMessage: null, }) + expect(retriedTask.detail?.message).toBeUndefined() const assignedRes = await app.request('/api/download-tasks?assignedTo=me&status=assigned', { headers: { Authorization: `Bearer ${createdDownloader.token}` }, diff --git a/server/services/downloads/core.ts b/server/services/downloads/core.ts index 019fc89c..07b27c33 100644 --- a/server/services/downloads/core.ts +++ b/server/services/downloads/core.ts @@ -6,7 +6,7 @@ import type { UpdateDownloaderInput, UpdateDownloadTaskInput, } from '@shared/schemas' -import type { Downloader, DownloadTask } from '@shared/types' +import type { Downloader, DownloadTask, DownloadTaskDetail } from '@shared/types' import { and, asc, count, desc, eq, inArray, like, sql } from 'drizzle-orm' import { nanoid } from 'nanoid' import { downloaders, downloadTasks } from '../../db/schema' @@ -30,6 +30,7 @@ const CANCELABLE_TASK_STATUSES = [ 'pausing', ] as const const TERMINAL_TASK_STATUSES = ['completed', 'failed', 'canceled'] as const +const EXECUTABLE_TASK_STATUSES = ['queued', 'assigned', 'downloading', 'uploading'] as const const RESTARTABLE_TASK_STATUSES = [ 'queued', 'assigned', @@ -141,6 +142,7 @@ export async function deleteDownloader(platform: Platform, id: string): Promise< uploadTokenJti: null, uploadTokenIssuedAt: null, uploadTokenExpiresAt: null, + detail: null, assignedAt: null, updatedAt: now, }) @@ -415,6 +417,7 @@ export async function updateDownloadTask( const nextFinishedAt = task.finishedAt ?? (input.status !== undefined && ['completed', 'failed', 'canceled'].includes(status) ? now : null) + const nextDetail = nextTaskDetail(task.detail, input.detail, status) await platform.db .update(downloadTasks) @@ -431,7 +434,7 @@ export async function updateDownloadTask( uploadBps: input.storageUploadBps ?? task.uploadBps, errorMessage: input.errorMessage === undefined ? task.errorMessage : input.errorMessage, resultObjectId: input.resultObjectId === undefined ? task.resultObjectId : input.resultObjectId, - detail: input.detail === undefined ? task.detail : JSON.stringify(input.detail), + detail: nextDetail, startedAt: task.startedAt ?? (status === 'downloading' ? now : null), finishedAt: nextFinishedAt, updatedAt: now, @@ -493,6 +496,7 @@ export async function performDownloadTaskAction( assignedAt: null, downloadBps: 0, uploadBps: 0, + detail: clearTaskDetailMessageJson(task.detail), updatedAt: now, }) .where(eq(downloadTasks.id, id)) @@ -540,6 +544,7 @@ export async function performDownloadTaskAction( uploadBps: 0, errorMessage: null, resultObjectId: null, + detail: clearTaskDetailMessageJson(task.detail), assignedAt: null, startedAt: null, finishedAt: null, @@ -588,6 +593,41 @@ export async function performDownloadTaskAction( throw new DownloadError('invalid_state') } +function nextTaskDetail( + current: string | null, + input: UpdateDownloadTaskInput['detail'], + status: string, +): string | null { + const detail = input === undefined ? parseTaskDetail(current) : input + if (!EXECUTABLE_TASK_STATUSES.includes(status as (typeof EXECUTABLE_TASK_STATUSES)[number])) { + return serializeTaskDetail(detail) + } + return serializeTaskDetail(clearTaskDetailMessage(detail)) +} + +function clearTaskDetailMessageJson(value: string | null): string | null { + return serializeTaskDetail(clearTaskDetailMessage(parseTaskDetail(value))) +} + +function clearTaskDetailMessage(detail: DownloadTaskDetail | null): DownloadTaskDetail | null { + if (!detail?.message) return detail + const { message: _message, ...rest } = detail + return Object.keys(rest).length > 0 ? rest : null +} + +function parseTaskDetail(value: string | null): DownloadTaskDetail | null { + if (!value) return null + try { + return JSON.parse(value) as DownloadTaskDetail + } catch { + return null + } +} + +function serializeTaskDetail(detail: DownloadTaskDetail | null | undefined): string | null { + return detail && Object.keys(detail).length > 0 ? JSON.stringify(detail) : null +} + export async function assertTaskUploadAllowed(platform: Platform, params: { taskId: string; downloaderId: string }) { const rows = await platform.db.select().from(downloadTasks).where(eq(downloadTasks.id, params.taskId)).limit(1) const task = rows[0]