diff --git a/downloader/internal/engine/aria2.go b/downloader/internal/engine/aria2.go index 8aaa0318..6620911b 100644 --- a/downloader/internal/engine/aria2.go +++ b/downloader/internal/engine/aria2.go @@ -36,7 +36,7 @@ var aria2StatusKeys = []string{ "downloadSpeed", "dir", "files", - "bitTorrent", + "bittorrent", "followedBy", "following", "belongsTo", diff --git a/downloader/internal/engine/engine_test.go b/downloader/internal/engine/engine_test.go index 06d0804d..f4c04d4b 100644 --- a/downloader/internal/engine/engine_test.go +++ b/downloader/internal/engine/engine_test.go @@ -116,6 +116,16 @@ func TestHTTPDownloadResumesExistingFile(t *testing.T) { } } +func TestAria2StatusKeysRequestBittorrentPayload(t *testing.T) { + keys := strings.Join(aria2StatusKeys, ",") + if !strings.Contains(keys, "bittorrent") { + t.Fatalf("expected aria2 status keys to request bittorrent payload, got %v", aria2StatusKeys) + } + if strings.Contains(keys, "bitTorrent") { + t.Fatalf("aria2 status key is case-sensitive; use bittorrent, got %v", aria2StatusKeys) + } +} + func TestQBittorrentCheckUsesWebAPIVersion(t *testing.T) { server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { if r.URL.Path != "/api/v2/app/version" { diff --git a/downloader/internal/worker/worker.go b/downloader/internal/worker/worker.go index 6fc66d27..769913bf 100644 --- a/downloader/internal/worker/worker.go +++ b/downloader/internal/worker/worker.go @@ -26,6 +26,13 @@ var errBillingPaused = errors.New("billing paused") var errTaskPausing = errors.New("task pausing") var errTaskCanceling = errors.New("task canceling") +type taskResumeStage int + +const ( + taskResumeDownload taskResumeStage = iota + taskResumeUpload +) + type Worker struct { cfg config.Config api apiClient @@ -175,7 +182,7 @@ 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 shouldRecoverBeforeDownload(task) { + if resumeStage(task) == taskResumeUpload { result, recovered, err := w.engine.Recover(ctx, task) if err != nil { msg := taskErrorMessage(err) @@ -268,20 +275,23 @@ func (w *Worker) process(ctx context.Context, task client.DownloadTask) { w.uploadAndComplete(ctx, log, task, result, currentDetail) } -func shouldRecoverBeforeDownload(task client.DownloadTask) bool { +func resumeStage(task client.DownloadTask) taskResumeStage { if task.Status == "uploading" { - return true + return taskResumeUpload } - if task.Status != "assigned" { - return false + if task.Status != "assigned" && task.Status != "running" { + return taskResumeDownload } if task.StorageUploadedBytes > 0 { - return true + return taskResumeUpload } - if task.Detail != nil && task.Detail.Phase == "uploading" { - return true + if task.Detail != nil && (task.Detail.Phase == "uploading" || task.Detail.Phase == "completed") { + return taskResumeUpload } - return task.TotalBytes != nil && *task.TotalBytes > 0 && task.DownloadedBytes >= *task.TotalBytes + if task.TotalBytes != nil && *task.TotalBytes > 0 && task.DownloadedBytes >= *task.TotalBytes { + return taskResumeUpload + } + return taskResumeDownload } func (w *Worker) uploadAndComplete( @@ -300,7 +310,22 @@ func (w *Worker) uploadAndComplete( if err != nil { msg := taskErrorMessage(err) log.Error("failed to upload result", "error", err) - if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "failed", ErrorMessage: &msg}); updateErr != nil { + downloadedBytes := result.Size + failedDetail := task.Detail + if failedDetail == nil { + failedDetail = &client.DownloadTaskDetail{} + } + failedDetail.Phase = "uploading" + failedDetail.PeerUploadBps = nil + if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{ + Status: "failed", + ErrorMessage: &msg, + DownloadedBytes: &downloadedBytes, + TotalBytes: &downloadedBytes, + DownloadBps: &zero, + StorageUploadBps: &zero, + Detail: failedDetail, + }); updateErr != nil { log.Error("failed to mark task failed", "error", updateErr) } return diff --git a/downloader/internal/worker/worker_test.go b/downloader/internal/worker/worker_test.go index 10a65f25..69f02ca1 100644 --- a/downloader/internal/worker/worker_test.go +++ b/downloader/internal/worker/worker_test.go @@ -4,6 +4,7 @@ import ( "context" "errors" "io" + "log/slog" "net/http" "net/http/httptest" "os" @@ -149,6 +150,112 @@ func TestCollectDirectoryEntriesSkipsDownloadSidecars(t *testing.T) { } } +func TestUploadFailurePersistsDownloadCheckpoint(t *testing.T) { + api := &recordingAPI{createFolderErr: errors.New("unauthorized")} + w := NewWithAPI(config.Config{}, api) + result := engine.Result{ + Path: t.TempDir(), + Name: "album", + Size: 1234, + IsDir: true, + } + + w.uploadAndComplete( + context.Background(), + slog.New(slog.NewTextHandler(io.Discard, nil)), + client.DownloadTask{ID: "task-1", Status: "running", UploadToken: "upload-token"}, + result, + nil, + ) + + if len(api.patches) < 2 { + t.Fatalf("expected uploading and failed updates, got %d", len(api.patches)) + } + failed := api.patches[len(api.patches)-1] + if failed.Status != "failed" { + t.Fatalf("expected failed status, got %q", failed.Status) + } + if failed.DownloadedBytes == nil || *failed.DownloadedBytes != result.Size { + t.Fatalf("expected downloaded bytes %d, got %#v", result.Size, failed.DownloadedBytes) + } + if failed.TotalBytes == nil || *failed.TotalBytes != result.Size { + t.Fatalf("expected total bytes %d, got %#v", result.Size, failed.TotalBytes) + } + if failed.Detail == nil || failed.Detail.Phase != "uploading" { + t.Fatalf("expected uploading detail phase, got %#v", failed.Detail) + } +} + +func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) { + payloadPath := writeTempFile(t, "downloaded payload") + payloadSize := int64(len("downloaded payload")) + uploadRequests := 0 + uploadServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + uploadRequests++ + if _, err := io.Copy(io.Discard, r.Body); err != nil { + t.Fatal(err) + } + w.WriteHeader(http.StatusOK) + })) + defer uploadServer.Close() + + api := &recordingAPI{ + createObjectDraft: client.ObjectDraft{ID: "object-1", Name: "payload.bin", UploadURL: uploadServer.URL}, + confirmErrs: []error{errors.New("unauthorized"), nil}, + } + eng := &recordingEngine{ + downloadResult: engine.Result{Path: payloadPath, Name: "payload.bin", Size: payloadSize}, + recoverResult: engine.Result{Path: payloadPath, Name: "payload.bin", Size: payloadSize}, + recovered: true, + } + + first := NewWithAPI(config.Config{}, api) + first.engine = eng + first.process(context.Background(), client.DownloadTask{ + ID: "task-1", + Status: "assigned", + SourceType: "http", + SourceURI: "https://example.com/payload.bin", + Name: "payload.bin", + UploadToken: "upload-token", + }) + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.DownloadedBytes == nil || *failed.DownloadedBytes != payloadSize { + t.Fatalf("expected failed task to persist downloaded bytes %d, got %#v", payloadSize, failed.DownloadedBytes) + } + if failed.Detail == nil || failed.Detail.Phase != "uploading" { + t.Fatalf("expected failed task to persist uploading phase, got %#v", failed.Detail) + } + + second := NewWithAPI(config.Config{}, api) + second.engine = eng + second.process(context.Background(), client.DownloadTask{ + ID: "task-1", + Status: "assigned", + SourceType: "http", + SourceURI: "https://example.com/payload.bin", + Name: "payload.bin", + DownloadedBytes: payloadSize, + TotalBytes: &payloadSize, + Detail: &client.DownloadTaskDetail{Phase: "uploading"}, + UploadToken: "upload-token", + }) + + if eng.downloadCalls != 1 { + t.Fatalf("expected retry to avoid a second download, got %d download calls", eng.downloadCalls) + } + if eng.recoverCalls != 1 { + t.Fatalf("expected retry to recover the completed download, got %d recover calls", eng.recoverCalls) + } + if uploadRequests != 2 { + t.Fatalf("expected both attempts to upload the local result, got %d upload requests", uploadRequests) + } + last := api.patches[len(api.patches)-1] + if last.Status != "completed" { + t.Fatalf("expected retry to complete task, got last patch %#v", last) + } +} + func TestUploadETARoundsRemainingSeconds(t *testing.T) { eta := uploadETA(&uploadProgress{uploaded: 25, totalBytes: 100}, 20) @@ -212,48 +319,58 @@ func TestTaskErrorMessageTruncatesToSchemaLimit(t *testing.T) { } } -func TestShouldRecoverBeforeDownload(t *testing.T) { +func TestResumeStage(t *testing.T) { total := int64(100) cases := []struct { name string task client.DownloadTask - want bool + want taskResumeStage }{ { name: "uploading status", task: client.DownloadTask{Status: "uploading"}, - want: true, + want: taskResumeUpload, }, { name: "assigned with upload bytes", task: client.DownloadTask{Status: "assigned", StorageUploadedBytes: 1}, - want: true, + want: taskResumeUpload, }, { name: "assigned with uploading phase", task: client.DownloadTask{Status: "assigned", Detail: &client.DownloadTaskDetail{Phase: "uploading"}}, - want: true, + want: taskResumeUpload, + }, + { + name: "assigned with completed phase", + task: client.DownloadTask{Status: "assigned", Detail: &client.DownloadTaskDetail{Phase: "completed"}}, + want: taskResumeUpload, }, { name: "assigned with completed download bytes", task: client.DownloadTask{Status: "assigned", DownloadedBytes: 100, TotalBytes: &total}, - want: true, + want: taskResumeUpload, }, { name: "assigned partial download", task: client.DownloadTask{Status: "assigned", DownloadedBytes: 99, TotalBytes: &total}, - want: false, + want: taskResumeDownload, }, { name: "running completed bytes", task: client.DownloadTask{Status: "running", DownloadedBytes: 100, TotalBytes: &total}, - want: false, + want: taskResumeUpload, + }, + { + name: "running partial download", + task: client.DownloadTask{Status: "running", DownloadedBytes: 99, TotalBytes: &total}, + want: taskResumeDownload, }, } for _, tc := range cases { t.Run(tc.name, func(t *testing.T) { - if got := shouldRecoverBeforeDownload(tc.task); got != tc.want { + if got := resumeStage(tc.task); got != tc.want { t.Fatalf("expected %v, got %v", tc.want, got) } }) @@ -404,3 +521,101 @@ func writeTempFile(t *testing.T, content string) string { func clientTask(id string) client.DownloadTask { return client.DownloadTask{ID: id, SourceType: "magnet"} } + +func lastPatchWithStatus(t *testing.T, patches []client.TaskPatch, status string) client.TaskPatch { + t.Helper() + for i := len(patches) - 1; i >= 0; i-- { + if patches[i].Status == status { + return patches[i] + } + } + t.Fatalf("expected patch with status %q in %#v", status, patches) + return client.TaskPatch{} +} + +type recordingEngine struct { + downloadResult engine.Result + recoverResult engine.Result + recovered bool + downloadCalls int + recoverCalls int +} + +func (e *recordingEngine) Name() string { + return "recording" +} + +func (e *recordingEngine) Capabilities() []string { + return []string{"http", "magnet", "torrent"} +} + +func (e *recordingEngine) Check(context.Context) error { + return nil +} + +func (e *recordingEngine) Recover(context.Context, client.DownloadTask) (engine.Result, bool, error) { + e.recoverCalls++ + return e.recoverResult, e.recovered, nil +} + +func (e *recordingEngine) Download(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error) { + e.downloadCalls++ + return e.downloadResult, nil +} + +type recordingAPI struct { + patches []client.TaskPatch + createFolderErr error + createObjectDraft client.ObjectDraft + confirmErrs []error +} + +func (a *recordingAPI) Heartbeat(context.Context, client.Heartbeat) error { + return nil +} + +func (a *recordingAPI) AssignedControlTasks(context.Context) ([]client.DownloadTask, error) { + return nil, nil +} + +func (a *recordingAPI) AssignedTasks(context.Context) ([]client.DownloadTask, error) { + return nil, nil +} + +func (a *recordingAPI) UpdateTask(_ context.Context, id string, patch client.TaskPatch) (client.DownloadTask, error) { + a.patches = append(a.patches, patch) + return client.DownloadTask{ID: id, Status: patch.Status, Detail: patch.Detail}, nil +} + +func (a *recordingAPI) CreateFolder(context.Context, string, string, string) (client.ObjectDraft, error) { + return client.ObjectDraft{}, a.createFolderErr +} + +func (a *recordingAPI) CreateObject(context.Context, string, string, int64, string) (client.ObjectDraft, error) { + return a.createObjectDraft, nil +} + +func (a *recordingAPI) ConfirmObject(context.Context, string, string) error { + if len(a.confirmErrs) > 0 { + err := a.confirmErrs[0] + a.confirmErrs = a.confirmErrs[1:] + return err + } + return nil +} + +func (a *recordingAPI) CreateObjectUploadSession(context.Context, string, string, int64) (client.ObjectUploadSession, error) { + return client.ObjectUploadSession{}, nil +} + +func (a *recordingAPI) PresignObjectUploadParts(context.Context, string, string, string, []int) ([]client.PresignedObjectUploadPart, error) { + return nil, nil +} + +func (a *recordingAPI) CompleteObjectUploadSession(context.Context, string, string, string, []client.CompletedObjectUploadPart) error { + return nil +} + +func (a *recordingAPI) AbortObjectUploadSession(context.Context, string, string, string) error { + return nil +} diff --git a/server/routes/download-tasks.integration.test.ts b/server/routes/download-tasks.integration.test.ts index ce689b53..20d32966 100644 --- a/server/routes/download-tasks.integration.test.ts +++ b/server/routes/download-tasks.integration.test.ts @@ -619,6 +619,97 @@ describe('Download tasks API integration', () => { await expect(deleteRes.json()).resolves.toEqual({ id: createdTask.id, deleted: true }) }) + it('preserves the completed download checkpoint when retrying an upload failure', async () => { + const { app, db } = await createTestApp({ DOWNLOAD_TOKEN_SECRET: 'test-download-token-secret' }) + await insertStorage(db) + + const createdDownloader = await registerDownloaderThroughDeviceLogin(app, 'retry-checkpoint-downloader') + const downloaderHeaders = { + Authorization: `Bearer ${createdDownloader.token}`, + 'Content-Type': 'application/json', + } + const heartbeatRes = await app.request('/api/downloader/heartbeat', { + method: 'POST', + headers: downloaderHeaders, + body: JSON.stringify({ ...heartbeat, currentTasks: 0 }), + }) + expect(heartbeatRes.status).toBe(200) + + const user = await authedHeaders(app, 'download-retry-checkpoint-user@example.com') + const createTaskRes = await app.request('/api/download-tasks', { + method: 'POST', + headers: { ...user, 'Content-Type': 'application/json' }, + body: JSON.stringify({ + source: { type: 'magnet', uri: 'magnet:?xt=urn:btih:abc123' }, + targetFolder: 'Media/Music', + }), + }) + expect(createTaskRes.status).toBe(201) + const createdTask = (await createTaskRes.json()) as { id: string; status: string } + expect(createdTask.status).toBe('assigned') + + const totalBytes = 4096 + const failedUploadRes = await app.request(`/api/download-tasks/${createdTask.id}`, { + method: 'PATCH', + headers: downloaderHeaders, + body: JSON.stringify({ + status: 'failed', + downloadedBytes: totalBytes, + totalBytes, + storageUploadedBytes: 1024, + errorMessage: 'confirm object failed', + detail: { engine: 'aria2', phase: 'uploading', infoHash: 'abc123' }, + }), + }) + expect(failedUploadRes.status).toBe(200) + await expect(failedUploadRes.json()).resolves.toMatchObject({ + status: 'failed', + downloadedBytes: totalBytes, + totalBytes, + storageUploadedBytes: 1024, + detail: { phase: 'uploading' }, + }) + + const retryRes = await app.request(`/api/download-tasks/${createdTask.id}/actions`, { + method: 'POST', + headers: { ...user, 'Content-Type': 'application/json' }, + body: JSON.stringify({ action: 'retry' }), + }) + expect(retryRes.status).toBe(200) + await expect(retryRes.json()).resolves.toMatchObject({ + status: 'assigned', + assignedDownloaderId: createdDownloader.downloader.id, + downloadedBytes: totalBytes, + totalBytes, + storageUploadedBytes: 1024, + detail: { phase: 'uploading' }, + errorMessage: null, + }) + + const assignedRes = await app.request('/api/download-tasks?assignedTo=me&status=assigned', { + headers: { Authorization: `Bearer ${createdDownloader.token}` }, + }) + expect(assignedRes.status).toBe(200) + const assigned = (await assignedRes.json()) as { + items: Array<{ + id: string + downloadedBytes: number + totalBytes: number + storageUploadedBytes: number + detail: { phase: string } + uploadToken: string + }> + } + const task = assigned.items.find((item) => item.id === createdTask.id) + expect(task).toMatchObject({ + downloadedBytes: totalBytes, + totalBytes, + storageUploadedBytes: 1024, + detail: { phase: 'uploading' }, + }) + expect(task?.uploadToken).toBeTruthy() + }) + it('uses transitional states for running task pause and cancel actions', async () => { const { app, db } = await createTestApp({ DOWNLOAD_TOKEN_SECRET: 'test-download-token-secret' }) await insertStorage(db) diff --git a/server/services/downloads/core.ts b/server/services/downloads/core.ts index 88f6ea44..72333997 100644 --- a/server/services/downloads/core.ts +++ b/server/services/downloads/core.ts @@ -517,14 +517,10 @@ export async function performDownloadTaskAction( uploadTokenHash: null, uploadTokenJti: null, uploadTokenExpiresAt: null, - downloadedBytes: 0, - uploadedBytes: 0, - totalBytes: null, downloadBps: 0, uploadBps: 0, errorMessage: null, resultObjectId: null, - detail: null, assignedAt: null, startedAt: null, finishedAt: null,