fix(downloader): report resumable state on shutdown

downloader/internal/engine/aria2.go

@@ -90,7 +90,7 @@ func (a Aria2) Start(ctx context.Context) (*exec.Cmd, error) {
 	if a.Secret != "" {
 		args = append(args, "--rpc-secret="+a.Secret)
 	}
-	cmd := exec.CommandContext(ctx, path, args...)
+	cmd := exec.Command(path, args...)
 	if err := cmd.Start(); err != nil {
 		return nil, err
 	}

downloader/internal/engine/qbittorrent.go

@@ -45,7 +45,7 @@ func (q QBittorrent) Start(ctx context.Context) (*exec.Cmd, error) {
 	if strings.Contains(filepathBase(path), "qbittorrent-nox") {
 		args = append(args, "--webui-port="+webURL.port)
 	}
-	cmd := exec.CommandContext(ctx, path, args...)
+	cmd := exec.Command(path, args...)
 	if err := cmd.Start(); err != nil {
 		return nil, err
 	}

downloader/internal/worker/worker.go

@@ -254,6 +254,15 @@ func (w *Worker) process(ctx context.Context, task client.DownloadTask) {
 				log.Info("task canceled by control action")
 				return
 			}
+			zero := int64(0)
+			if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{
+				Status:           "assigned",
+				DownloadBps:      &zero,
+				StorageUploadBps: &zero,
+				Detail:           currentDetail,
+			}); updateErr != nil {
+				log.Error("failed to mark task resumable after shutdown", "error", updateErr)
+			}
 			log.Info("task stopped by context cancellation")
 			return
 		}
@@ -313,9 +322,29 @@ func (w *Worker) uploadAndComplete(
 	task.Detail = currentDetail
 	resultObjectID, err := w.uploadResult(ctx, log, task, result)
 	if err != nil {
+		downloadedBytes := result.Size
+		if errors.Is(err, context.Canceled) {
+			uploadingDetail := task.Detail
+			if uploadingDetail == nil {
+				uploadingDetail = &client.DownloadTaskDetail{}
+			}
+			uploadingDetail.Phase = "uploading"
+			uploadingDetail.PeerUploadBps = nil
+			if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{
+				Status:           "uploading",
+				DownloadedBytes:  &downloadedBytes,
+				TotalBytes:       &downloadedBytes,
+				DownloadBps:      &zero,
+				StorageUploadBps: &zero,
+				Detail:           uploadingDetail,
+			}); updateErr != nil {
+				log.Error("failed to mark task uploading after shutdown", "error", updateErr)
+			}
+			log.Info("task upload stopped by context cancellation")
+			return
+		}
 		msg := taskErrorMessage(err)
 		log.Error("failed to upload result", "error", err)
-		downloadedBytes := result.Size
 		failedDetail := task.Detail
 		if failedDetail == nil {
 			failedDetail = &client.DownloadTaskDetail{}

downloader/internal/worker/worker_test.go

@@ -256,6 +256,46 @@ func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) {
 	}
 }
 
+func TestDownloadShutdownMarksTaskAssigned(t *testing.T) {
+	api := &recordingAPI{}
+	eng := &recordingEngine{downloadErr: context.Canceled}
+	w := NewWithAPI(config.Config{}, api)
+	w.engine = eng
+
+	w.process(context.Background(), client.DownloadTask{ID: "task-1", Status: "running"})
+
+	patch := lastPatchWithStatus(t, api.patches, "assigned")
+	if patch.DownloadBps == nil || *patch.DownloadBps != 0 {
+		t.Fatalf("expected download speed to be reset, got %#v", patch.DownloadBps)
+	}
+}
+
+func TestUploadShutdownKeepsTaskUploading(t *testing.T) {
+	payloadPath := writeTempFile(t, "downloaded payload")
+	api := &recordingAPI{
+		createObjectDraft: client.ObjectDraft{ID: "object-1", Name: "payload.bin", UploadURL: "http://127.0.0.1:1"},
+	}
+	w := NewWithAPI(config.Config{}, api)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	w.uploadAndComplete(
+		ctx,
+		slog.New(slog.NewTextHandler(io.Discard, nil)),
+		client.DownloadTask{ID: "task-1", Status: "running", UploadToken: "upload-token"},
+		engine.Result{Path: payloadPath, Name: "payload.bin", Size: int64(len("downloaded payload"))},
+		nil,
+	)
+
+	patch := lastPatchWithStatus(t, api.patches, "uploading")
+	if patch.DownloadedBytes == nil || *patch.DownloadedBytes != int64(len("downloaded payload")) {
+		t.Fatalf("expected upload shutdown to preserve downloaded checkpoint, got %#v", patch.DownloadedBytes)
+	}
+	if _, ok := findPatchWithStatus(api.patches, "failed"); ok {
+		t.Fatalf("expected upload shutdown not to mark failed, got %#v", api.patches)
+	}
+}
+
 func TestUploadETARoundsRemainingSeconds(t *testing.T) {
 	eta := uploadETA(&uploadProgress{uploaded: 25, totalBytes: 100}, 20)
 
@@ -609,17 +649,25 @@ func clientTask(id string) client.DownloadTask {
 
 func lastPatchWithStatus(t *testing.T, patches []client.TaskPatch, status string) client.TaskPatch {
 	t.Helper()
+	patch, ok := findPatchWithStatus(patches, status)
+	if !ok {
+		t.Fatalf("expected patch with status %q in %#v", status, patches)
+	}
+	return patch
+}
+
+func findPatchWithStatus(patches []client.TaskPatch, status string) (client.TaskPatch, bool) {
 	for i := len(patches) - 1; i >= 0; i-- {
 		if patches[i].Status == status {
-			return patches[i]
+			return patches[i], true
 		}
 	}
-	t.Fatalf("expected patch with status %q in %#v", status, patches)
-	return client.TaskPatch{}
+	return client.TaskPatch{}, false
 }
 
 type recordingEngine struct {
 	downloadResult engine.Result
+	downloadErr    error
 	recoverResult  engine.Result
 	recovered      bool
 	restoreSeed    *engine.Seed
@@ -652,7 +700,7 @@ func (e *recordingEngine) RestoreSeed(context.Context, engine.SeedRef) (*engine.
 
 func (e *recordingEngine) Download(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error) {
 	e.downloadCalls++
-	return e.downloadResult, nil
+	return e.downloadResult, e.downloadErr
 }
 
 type recordingAPI struct {