mirror of
https://github.com/saltbo/zpan.git
synced 2026-08-28 15:51:29 +08:00
fix(downloader): preserve retry recovery checkpoints
This commit is contained in:
saltbo
@@ -36,7 +36,7 @@ var aria2StatusKeys = []string{
|
||||
"downloadSpeed",
|
||||
"dir",
|
||||
"files",
|
||||
"bitTorrent",
|
||||
"bittorrent",
|
||||
"followedBy",
|
||||
"following",
|
||||
"belongsTo",
|
||||
|
||||
@@ -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" {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user