From 7cfbbf77b7a3bf4d7094df384de9d07d8d832aaf Mon Sep 17 00:00:00 2001 From: "agent-kanban-local[bot]" <292844359+agent-kanban-local[bot]@users.noreply.github.com> Date: Wed, 24 Jun 2026 03:14:38 -0400 Subject: [PATCH] fix: clean downloader terminal artifacts (#477) * fix: clean downloader terminal artifacts Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29 * test: tolerate stale cloud license cleanup Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29 * test: retry transient pairing poll failures Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29 * fix: make suspended downloader cleanup idempotent Agent-Profile: https://agent-kanban.dev/agents/57ed5bcf43079e29 --------- Co-authored-by: Jordan Park --- cmd/internal/worker/disk_unix.go | 41 ++ cmd/internal/worker/disk_unix_test.go | 25 ++ cmd/internal/worker/disk_windows.go | 45 ++ cmd/internal/worker/seeds.go | 14 +- cmd/internal/worker/worker.go | 110 ++++- cmd/internal/worker/worker_test.go | 576 ++++++++++++++++++++++++-- e2e/helpers.ts | 17 +- 7 files changed, 780 insertions(+), 48 deletions(-) create mode 100644 cmd/internal/worker/disk_unix.go create mode 100644 cmd/internal/worker/disk_unix_test.go create mode 100644 cmd/internal/worker/disk_windows.go diff --git a/cmd/internal/worker/disk_unix.go b/cmd/internal/worker/disk_unix.go new file mode 100644 index 00000000..ff263bd0 --- /dev/null +++ b/cmd/internal/worker/disk_unix.go @@ -0,0 +1,41 @@ +//go:build !windows + +package worker + +import ( + "os" + "path/filepath" + + "golang.org/x/sys/unix" +) + +func freeDiskBytes(path string) (int64, error) { + statPath, err := existingStatPath(path) + if err != nil { + return 0, err + } + var stat unix.Statfs_t + if err := unix.Statfs(statPath, &stat); err != nil { + return 0, err + } + return int64(stat.Bavail) * int64(stat.Bsize), nil +} + +func existingStatPath(path string) (string, error) { + if path == "" { + return ".", nil + } + path = filepath.Clean(path) + for { + if _, err := os.Stat(path); err == nil { + return path, nil + } else if !os.IsNotExist(err) { + return "", err + } + parent := filepath.Dir(path) + if parent == path { + return "", os.ErrNotExist + } + path = parent + } +} diff --git a/cmd/internal/worker/disk_unix_test.go b/cmd/internal/worker/disk_unix_test.go new file mode 100644 index 00000000..9b649b17 --- /dev/null +++ b/cmd/internal/worker/disk_unix_test.go @@ -0,0 +1,25 @@ +//go:build !windows + +package worker + +import ( + "testing" + + "github.com/saltbo/zpan/internal/config" + "golang.org/x/sys/unix" +) + +func TestHeartbeatReportsDownloadDirFreeDiskExactly(t *testing.T) { + downloadDir := t.TempDir() + var stat unix.Statfs_t + if err := unix.Statfs(downloadDir, &stat); err != nil { + t.Fatalf("statfs %s: %v", downloadDir, err) + } + want := int64(stat.Bavail) * int64(stat.Bsize) + + w := NewWithAPI(config.Config{DownloadDir: downloadDir}, &recordingAPI{}) + + if got := w.heartbeat().FreeDiskBytes; got != want { + t.Fatalf("expected heartbeat free disk %d, got %d", want, got) + } +} diff --git a/cmd/internal/worker/disk_windows.go b/cmd/internal/worker/disk_windows.go new file mode 100644 index 00000000..6ac16856 --- /dev/null +++ b/cmd/internal/worker/disk_windows.go @@ -0,0 +1,45 @@ +//go:build windows + +package worker + +import ( + "os" + "path/filepath" + + "golang.org/x/sys/windows" +) + +func freeDiskBytes(path string) (int64, error) { + statPath, err := existingStatPath(path) + if err != nil { + return 0, err + } + ptr, err := windows.UTF16PtrFromString(statPath) + if err != nil { + return 0, err + } + var freeBytes uint64 + if err := windows.GetDiskFreeSpaceEx(ptr, &freeBytes, nil, nil); err != nil { + return 0, err + } + return int64(freeBytes), nil +} + +func existingStatPath(path string) (string, error) { + if path == "" { + return ".", nil + } + path = filepath.Clean(path) + for { + if _, err := os.Stat(path); err == nil { + return path, nil + } else if !os.IsNotExist(err) { + return "", err + } + parent := filepath.Dir(path) + if parent == path { + return "", os.ErrNotExist + } + path = parent + } +} diff --git a/cmd/internal/worker/seeds.go b/cmd/internal/worker/seeds.go index 14060dfe..864c95fd 100644 --- a/cmd/internal/worker/seeds.go +++ b/cmd/internal/worker/seeds.go @@ -30,11 +30,21 @@ type retainedSeed struct { cleanup func(context.Context) error } -func cleanupDownloadedResult(ctx context.Context, result engine.Result) error { +func cleanupDownloadedResult(ctx context.Context, task client.DownloadTask, result engine.Result) error { if result.Seed != nil && result.Seed.Cleanup != nil { return result.Seed.Cleanup(ctx) } - return os.RemoveAll(result.Path) + parent := filepath.Dir(result.Path) + if filepath.Base(parent) == task.ID { + return os.RemoveAll(parent) + } + if result.IsDir { + return os.RemoveAll(result.Path) + } + if err := os.Remove(result.Path); err != nil && !os.IsNotExist(err) { + return err + } + return nil } func (w *Worker) retainSeed(task client.DownloadTask, result engine.Result, log *slog.Logger) bool { diff --git a/cmd/internal/worker/worker.go b/cmd/internal/worker/worker.go index c3e9a015..1daa2433 100644 --- a/cmd/internal/worker/worker.go +++ b/cmd/internal/worker/worker.go @@ -23,6 +23,7 @@ import ( const Version = "0.1.0" const maxTaskErrorMessageLength = 1000 +const localResultRemovedRuntimeState = "local_result_removed" var errTaskPausing = errors.New("task pausing") var errTaskCanceling = errors.New("task canceling") @@ -297,6 +298,7 @@ func (w *Worker) downloadThenUpload( return } if errors.Is(context.Cause(ctx), errTaskCanceling) { + w.cleanupTerminalTask(context.WithoutCancel(ctx), log, task, "canceled") if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{Status: "canceled"}); updateErr != nil { log.Error("failed to mark task canceled", "error", updateErr) } @@ -306,6 +308,7 @@ func (w *Worker) downloadThenUpload( if errors.Is(context.Cause(ctx), errTaskSuspended) { // The server already moved the task to suspended (billing); the // poll told us to stop. Don't touch its status. + w.cleanupSuspendedTask(context.WithoutCancel(ctx), log, task) log.Info("task stopped because it was suspended") return } @@ -326,6 +329,7 @@ func (w *Worker) downloadThenUpload( if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "failed", ErrorMessage: &msg}); updateErr != nil { log.Error("failed to mark task failed", "error", updateErr) } + w.cleanupTerminalTask(context.WithoutCancel(ctx), log, task, "failed") return } @@ -393,13 +397,16 @@ func nextTaskWorkStage(task client.DownloadTask) taskWorkStage { if task.State() == "uploading" { return taskWorkStageUploadExistingResult } + runtime := task.Runtime() + if runtime != nil && runtime.State == localResultRemovedRuntimeState { + return taskWorkStageDownload + } if task.State() != "assigned" && task.State() != "downloading" && task.State() != "interrupted" { return taskWorkStageDownload } if task.Status.Progress.Upload.Bytes > 0 { return taskWorkStageUploadExistingResult } - runtime := task.Runtime() if runtime != nil && (runtime.Phase == "uploading" || runtime.Phase == "completed") { return taskWorkStageUploadExistingResult } @@ -477,6 +484,41 @@ func (w *Worker) resetRuntimeTask(ctx context.Context, task client.DownloadTask, return nil } +func (w *Worker) cleanupTerminalTask(ctx context.Context, log *slog.Logger, task client.DownloadTask, reason string) bool { + w.cleanupRetainedSeedForTask(ctx, task.ID, reason) + if w.engine == nil { + log.Warn("downloader engine is unavailable for terminal cleanup", "reason", reason) + return false + } + resetter, ok := w.engine.(engine.TaskResetter) + if !ok { + log.Warn("downloader engine does not support terminal cleanup", "engine", w.engine.Name(), "reason", reason) + return false + } + if err := resetter.ResetTask(ctx, task); err != nil { + log.Warn("failed to clean terminal downloader task", "reason", reason, "error", err) + return false + } + log.Info("cleaned terminal downloader task", "reason", reason) + return true +} + +func (w *Worker) cleanupSuspendedTask(ctx context.Context, log *slog.Logger, task client.DownloadTask) { + runtime := task.Runtime() + if runtime != nil && runtime.State == localResultRemovedRuntimeState { + log.Debug("suspended task cleanup already recorded") + return + } + if !w.cleanupTerminalTask(ctx, log, task, "suspended") { + return + } + if _, err := w.updateTask(ctx, task.ID, client.TaskPatch{ + Runtime: localResultRemovedRuntime(runtime), + }); err != nil { + log.Error("failed to record suspended task cleanup", "error", err) + } +} + func (w *Worker) uploadAndComplete( ctx context.Context, log *slog.Logger, @@ -498,6 +540,38 @@ func (w *Worker) uploadAndComplete( if err != nil { downloadedBytes := result.Size if errors.Is(err, context.Canceled) { + if errors.Is(context.Cause(ctx), errTaskPausing) { + if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{Status: "paused"}); updateErr != nil { + log.Error("failed to mark task paused during upload", "error", updateErr) + } + log.Info("task upload paused by control action") + return + } + if errors.Is(context.Cause(ctx), errTaskCanceling) { + if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil { + log.Warn("failed to remove canceled local downloaded result", "path", result.Path, "error", cleanupErr) + } + if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{ + Status: "canceled", + Runtime: localResultRemovedRuntime(currentDetail), + }); updateErr != nil { + log.Error("failed to mark task canceled during upload", "error", updateErr) + } + log.Info("task upload canceled by control action") + return + } + if errors.Is(context.Cause(ctx), errTaskSuspended) { + if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil { + log.Warn("failed to remove suspended local downloaded result", "path", result.Path, "error", cleanupErr) + } + if _, updateErr := w.updateTask(context.WithoutCancel(ctx), task.ID, client.TaskPatch{ + Runtime: localResultRemovedRuntime(currentDetail), + }); updateErr != nil { + log.Error("failed to record suspended upload cleanup", "error", updateErr) + } + log.Info("task upload stopped because it was suspended") + return + } uploadingDetail := currentDetail if uploadingDetail == nil { uploadingDetail = &client.DownloadTaskRuntime{} @@ -519,12 +593,7 @@ func (w *Worker) uploadAndComplete( } msg := taskErrorMessage(err) log.Error("failed to upload result", "error", err) - failedDetail := currentDetail - if failedDetail == nil { - failedDetail = &client.DownloadTaskRuntime{} - } - failedDetail.Phase = "uploading" - failedDetail.Seeding = nil + failedDetail := localResultRemovedRuntime(currentDetail) if _, updateErr := w.updateTask(ctx, task.ID, client.TaskPatch{ Status: "failed", ErrorMessage: &msg, @@ -536,6 +605,9 @@ func (w *Worker) uploadAndComplete( }); updateErr != nil { log.Error("failed to mark task failed", "error", updateErr) } + if cleanupErr := cleanupDownloadedResult(context.WithoutCancel(ctx), task, result); cleanupErr != nil { + log.Warn("failed to remove failed local downloaded result", "path", result.Path, "error", cleanupErr) + } return } uploadedBytes := result.Size @@ -564,7 +636,7 @@ func (w *Worker) uploadAndComplete( w.cleanupRetainedSeeds(ctx) return } - if err := cleanupDownloadedResult(ctx, result); err != nil { + if err := cleanupDownloadedResult(ctx, task, result); err != nil { log.Warn("failed to remove local downloaded result", "path", result.Path, "error", err) } } @@ -595,6 +667,17 @@ func interruptedRuntime(runtime *client.DownloadTaskRuntime) *client.DownloadTas return runtime } +func localResultRemovedRuntime(runtime *client.DownloadTaskRuntime) *client.DownloadTaskRuntime { + if runtime == nil { + runtime = &client.DownloadTaskRuntime{} + } + runtime.State = localResultRemovedRuntimeState + runtime.Phase = "error" + runtime.Seeding = nil + runtime.ETASeconds = nil + return runtime +} + func downloadProgressPatch(downloaded int64, total *int64, bps int64) *client.DownloadTaskProgressPatch { return &client.DownloadTaskProgressPatch{Download: transferProgress(downloaded, total, bps)} } @@ -796,11 +879,16 @@ func (w *Worker) ackStoppedControlTask(ctx context.Context, task client.Download return } if task.State() == "canceling" { + w.cleanupTerminalTask(ctx, log, task, "canceled") if _, err := w.updateTask(ctx, task.ID, client.TaskPatch{Status: "canceled"}); err != nil { log.Error("failed to acknowledge canceled task without local process", "error", err) return } log.Info("acknowledged canceled task without local process") + return + } + if task.State() == "suspended" { + w.cleanupSuspendedTask(ctx, log, task) } } @@ -812,6 +900,10 @@ func (w *Worker) heartbeat() client.Heartbeat { capabilities = w.engine.Capabilities() } speeds := w.currentTransferSpeeds() + freeDiskBytes, err := freeDiskBytes(w.cfg.DownloadDir) + if err != nil { + w.logger.Warn("failed to inspect downloader free disk space", "download_dir", w.cfg.DownloadDir, "error", err) + } return client.Heartbeat{ Version: Version, Hostname: host.DownloaderHostname(), @@ -823,7 +915,7 @@ func (w *Worker) heartbeat() client.Heartbeat { CurrentTasks: w.currentTasks(), DownloadBps: speeds.downloadBps, UploadBps: speeds.uploadBps, - FreeDiskBytes: 0, + FreeDiskBytes: freeDiskBytes, } } diff --git a/cmd/internal/worker/worker_test.go b/cmd/internal/worker/worker_test.go index 7943b855..3cdd807e 100644 --- a/cmd/internal/worker/worker_test.go +++ b/cmd/internal/worker/worker_test.go @@ -64,6 +64,234 @@ func TestDownloadThenUploadStopsWhenSuspendedAtStart(t *testing.T) { } } +func TestCanceledDownloadCleansRuntimeAndMarksCanceled(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{downloadErr: context.Canceled} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(errTaskCanceling) + + w.downloadThenUpload(ctx, w.logger, clientTaskWithStatus("task-1", "downloading"), nil) + + if eng.resetCalls != 1 { + t.Fatalf("expected canceled task cleanup to reset runtime once, got %d", eng.resetCalls) + } + patch := lastPatchWithStatus(t, api.patches, "canceled") + if patch.State() != "canceled" { + t.Fatalf("expected canceled patch, got %#v", patch) + } +} + +func TestSuspendedDownloadCleansRuntimeWithoutStatusChange(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{downloadErr: context.Canceled} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(errTaskSuspended) + + w.downloadThenUpload(ctx, w.logger, clientTaskWithStatus("task-1", "downloading"), nil) + + if eng.resetCalls != 1 { + t.Fatalf("expected suspended task cleanup to reset runtime once, got %d", eng.resetCalls) + } + if _, ok := findPatchWithStatus(api.patches, "suspended"); ok { + t.Fatalf("expected worker not to overwrite server-owned suspended status, got %#v", api.patches) + } + patch := api.patches[len(api.patches)-1] + if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState { + t.Fatalf("expected suspended cleanup marker, got %#v", patch.Runtime) + } +} + +func TestTickSuspendedControlTaskCleansRuntimeOnlyOnce(t *testing.T) { + api := &recordingAPI{ + controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")}, + } + eng := &recordingEngine{} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + + if err := w.tick(context.Background()); err != nil { + t.Fatalf("first tick: %v", err) + } + if eng.resetCalls != 1 { + t.Fatalf("expected first suspended control poll to clean once, got %d", eng.resetCalls) + } + if len(api.patches) != 1 { + t.Fatalf("expected first suspended control cleanup to record runtime once, got %#v", api.patches) + } + patch := api.patches[0] + if patch.State() != "" { + t.Fatalf("expected suspended control cleanup to preserve server-owned status, got %#v", patch) + } + if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState { + t.Fatalf("expected suspended control cleanup marker, got %#v", patch.Runtime) + } + + if err := w.tick(context.Background()); err != nil { + t.Fatalf("second tick: %v", err) + } + if eng.resetCalls != 1 { + t.Fatalf("expected repeated suspended control polls to avoid duplicate cleanup, got %d", eng.resetCalls) + } + if len(api.patches) != 1 { + t.Fatalf("expected repeated suspended control polls not to rewrite cleanup marker, got %#v", api.patches) + } + if got := api.controlTasks[0].State(); got != "suspended" { + t.Fatalf("expected recorded control task status to stay suspended, got %q", got) + } +} + +func TestSuspendedControlTaskResumeAfterCleanupRedownloads(t *testing.T) { + api := &recordingAPI{ + controlTasks: []client.DownloadTask{clientTaskWithStatus("task-1", "suspended")}, + } + eng := &recordingEngine{} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + + if err := w.tick(context.Background()); err != nil { + t.Fatalf("tick: %v", err) + } + marker := api.controlTasks[0].Runtime() + if marker == nil || marker.State != localResultRemovedRuntimeState { + t.Fatalf("expected suspended control task to record cleanup marker, got %#v", marker) + } + + total := int64(100) + eng.downloadErr = errors.New("redownload missing local result") + w.process(context.Background(), withRuntime( + withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), total, &total), + marker, + )) + + if eng.inspectCalls != 0 { + t.Fatalf("expected cleaned suspended result to skip runtime upload inspection, got %d inspect calls", eng.inspectCalls) + } + if eng.downloadCalls != 1 { + t.Fatalf("expected cleaned suspended result to resume via download path, got %d download calls", eng.downloadCalls) + } + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "redownload missing local result") { + t.Fatalf("expected resumed redownload failure to be reported, got %#v", failed.ErrorMessage) + } +} + +func TestFailedDownloadCleansRuntimeAndMarksFailed(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{downloadErr: errors.New("disk write failed")} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + + w.downloadThenUpload(context.Background(), w.logger, clientTaskWithStatus("task-1", "downloading"), nil) + + if eng.resetCalls != 1 { + t.Fatalf("expected failed task cleanup to reset runtime once, got %d", eng.resetCalls) + } + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "disk write failed") { + t.Fatalf("expected failure message to be reported, got %#v", failed.ErrorMessage) + } +} + +func TestTerminalDownloadStopsCleanPartialFiles(t *testing.T) { + cases := []struct { + name string + cancelCause error + failedMessage string + wantStatus string + requireStop bool + }{ + { + name: "canceled", + cancelCause: errTaskCanceling, + wantStatus: "canceled", + }, + { + name: "suspended", + cancelCause: errTaskSuspended, + requireStop: true, + }, + { + name: "failed", + failedMessage: "disk write failed", + wantStatus: "failed", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + api := &recordingAPI{} + downloadDir := t.TempDir() + partialPath := filepath.Join(downloadDir, "task-1", "payload.bin") + ready := make(chan struct{}, 1) + eng := &recordingEngine{ + downloadFunc: func(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) { + if err := os.MkdirAll(filepath.Dir(partialPath), 0o755); err != nil { + return engine.Result{}, err + } + if err := os.WriteFile(partialPath, []byte("partial"), 0o644); err != nil { + return engine.Result{}, err + } + ready <- struct{}{} + if tc.failedMessage != "" { + return engine.Result{}, errors.New(tc.failedMessage) + } + <-ctx.Done() + return engine.Result{}, ctx.Err() + }, + resetTaskFn: func(context.Context, client.DownloadTask) error { + return os.RemoveAll(filepath.Join(downloadDir, "task-1")) + }, + } + w := NewWithAPI(config.Config{DownloadDir: downloadDir}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + + task := clientHTTPTask("task-1", "downloading", "https://example.com/payload.bin", "payload.bin") + if tc.cancelCause != nil { + ctx, cancel := context.WithCancelCause(context.Background()) + done := make(chan struct{}) + go func() { + w.downloadThenUpload(ctx, w.logger, task, nil) + close(done) + }() + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for partial download") + } + cancel(tc.cancelCause) + waitForWorkerTestCompletion(t, done) + } else { + w.downloadThenUpload(context.Background(), w.logger, task, nil) + } + + if tc.wantStatus != "" { + lastPatchWithStatus(t, api.patches, tc.wantStatus) + } + taskDir := filepath.Join(downloadDir, task.ID) + if _, err := os.Stat(taskDir); !os.IsNotExist(err) { + t.Fatalf("expected %s cleanup after %s stop, got err=%v", taskDir, tc.name, err) + } + if tc.requireStop { + for _, forbidden := range []string{"failed", "interrupted", "canceled", "paused"} { + if _, ok := findPatchWithStatus(api.patches, forbidden); ok { + t.Fatalf("expected suspended stop not to emit %q, got %#v", forbidden, api.patches) + } + } + } + }) + } +} + func TestWatchEngineProcessFatalOnUnexpectedExit(t *testing.T) { w := NewWithAPI(config.Config{}, nil) w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) @@ -233,7 +461,7 @@ func TestReconcileEngineSeedsAdoptsUntrackedOrphans(t *testing.T) { } func TestHeartbeatReportsAggregateTransferSpeeds(t *testing.T) { - w := NewWithAPI(config.Config{Engine: "auto", MaxConcurrentTasks: 5}, &recordingAPI{}) + w := NewWithAPI(config.Config{Engine: "auto", MaxConcurrentTasks: 5, DownloadDir: t.TempDir()}, &recordingAPI{}) if _, ok := w.startTask(context.Background(), "task-1"); !ok { t.Fatal("expected task-1 to start") @@ -251,6 +479,9 @@ func TestHeartbeatReportsAggregateTransferSpeeds(t *testing.T) { if heartbeat.DownloadBps != 400 || heartbeat.UploadBps != 60 { t.Fatalf("expected aggregate speeds 400/60, got %d/%d", heartbeat.DownloadBps, heartbeat.UploadBps) } + if heartbeat.FreeDiskBytes <= 0 { + t.Fatalf("expected heartbeat to report free disk bytes, got %d", heartbeat.FreeDiskBytes) + } w.finish("task-1") heartbeat = w.heartbeat() @@ -445,11 +676,39 @@ func TestCollectDirectoryEntriesSkipsDownloadSidecars(t *testing.T) { } } +func TestCleanupDownloadedResultRemovesTaskDirForNestedDirectoryResult(t *testing.T) { + downloadDir := t.TempDir() + taskDir := filepath.Join(downloadDir, "task-1") + resultDir := filepath.Join(taskDir, "payload") + if err := os.MkdirAll(resultDir, 0o755); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(resultDir, "file.txt"), []byte("payload"), 0o644); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(filepath.Join(taskDir, "payload.torrent"), []byte("sidecar"), 0o644); err != nil { + t.Fatal(err) + } + + err := cleanupDownloadedResult(context.Background(), clientTaskWithStatus("task-1", "downloading"), engine.Result{ + Path: resultDir, + Name: "payload", + IsDir: true, + }) + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(taskDir); !os.IsNotExist(err) { + t.Fatalf("expected nested directory result cleanup to remove task dir, stat err=%v", err) + } +} + func TestUploadFailurePersistsDownloadCheckpoint(t *testing.T) { api := &recordingAPI{createFolderErr: errors.New("unauthorized")} w := NewWithAPI(config.Config{}, api) + resultPath := t.TempDir() result := engine.Result{ - Path: t.TempDir(), + Path: resultPath, Name: "album", Size: 1234, IsDir: true, @@ -476,12 +735,15 @@ func TestUploadFailurePersistsDownloadCheckpoint(t *testing.T) { if failed.Progress.Download.TotalBytes == nil || *failed.Progress.Download.TotalBytes != result.Size { t.Fatalf("expected total bytes %d, got %#v", result.Size, failed.Progress.Download.TotalBytes) } - if failed.Runtime == nil || failed.Runtime.Phase != "uploading" { - t.Fatalf("expected uploading detail phase, got %#v", failed.Runtime) + if failed.Runtime == nil || failed.Runtime.Phase != "error" || failed.Runtime.State != localResultRemovedRuntimeState { + t.Fatalf("expected local-result-removed runtime, got %#v", failed.Runtime) + } + if _, err := os.Stat(resultPath); !os.IsNotExist(err) { + t.Fatalf("expected upload failure to remove local result path, stat err=%v", err) } } -func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) { +func TestWorkerLifecycleUploadFailureCleansLocalResult(t *testing.T) { payloadPath := writeTempFile(t, "downloaded payload") payloadSize := int64(len("downloaded payload")) uploadRequests := 0 @@ -515,33 +777,18 @@ func TestWorkerLifecycleRetriesUploadWithoutRedownloading(t *testing.T) { if failed.Progress == nil || failed.Progress.Download == nil || failed.Progress.Download.Bytes != payloadSize { t.Fatalf("expected failed task to persist downloaded bytes %d, got %#v", payloadSize, failed.Progress) } - if failed.Runtime == nil || failed.Runtime.Phase != "uploading" { - t.Fatalf("expected failed task to persist uploading phase, got %#v", failed.Runtime) + if failed.Runtime == nil || failed.Runtime.State != localResultRemovedRuntimeState { + t.Fatalf("expected failed task to mark local result removed, got %#v", failed.Runtime) } - - second := NewWithAPI(config.Config{}, api) - second.engine = eng - second.process(context.Background(), withRuntime( - withDownloadCheckpoint(clientHTTPTask("task-1", "assigned", "https://example.com/payload.bin", "payload.bin"), payloadSize, &payloadSize), - &client.DownloadTaskRuntime{Phase: "uploading"}, - )) - - if eng.downloadCalls != 1 { - t.Fatalf("expected retry to avoid a second download, got %d download calls", eng.downloadCalls) + if uploadRequests != 1 { + t.Fatalf("expected one upload attempt, got %d", uploadRequests) } - if eng.inspectCalls != 1 { - t.Fatalf("expected retry to inspect the runtime task, got %d inspect calls", eng.inspectCalls) - } - 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.State() != "completed" { - t.Fatalf("expected retry to complete task, got last patch %#v", last) + if _, err := os.Stat(payloadPath); !os.IsNotExist(err) { + t.Fatalf("expected failed upload to remove local payload, stat err=%v", err) } } -func TestWorkerLifecycleRetriesHTTPUploadFromCheckpointWithoutRedownloading(t *testing.T) { +func TestWorkerLifecycleHTTPUploadFailureCleansLocalResult(t *testing.T) { payload := "downloaded payload" downloadRequests := 0 downloadServer := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { @@ -579,19 +826,26 @@ func TestWorkerLifecycleRetriesHTTPUploadFromCheckpointWithoutRedownloading(t *t if downloadRequests != 1 { t.Fatalf("expected initial attempt to download once, got %d requests", downloadRequests) } + if uploadRequests != 1 { + t.Fatalf("expected one upload attempt, got %d", uploadRequests) + } + if _, err := os.Stat(filepath.Join(downloadDir, "task-1")); !os.IsNotExist(err) { + t.Fatalf("expected failed upload to remove local task directory, stat err=%v", err) + } second := NewWithAPI(config.Config{}, api) second.engine = engine.HTTP{Dir: downloadDir} + failedRuntime := failed.Runtime second.process(context.Background(), withRuntime( withDownloadCheckpoint(clientHTTPTask("task-1", "assigned", downloadServer.URL+"/payload.bin", "payload.bin"), payloadSize, &payloadSize), - &client.DownloadTaskRuntime{Phase: "uploading"}, + failedRuntime, )) - if downloadRequests != 1 { - t.Fatalf("expected retry not to request download source again, got %d requests", downloadRequests) + if downloadRequests != 2 { + t.Fatalf("expected retry after cleanup to redownload, got %d requests", downloadRequests) } if uploadRequests != 2 { - t.Fatalf("expected both attempts to upload the local file, got %d upload requests", uploadRequests) + t.Fatalf("expected retry to upload redownloaded file, got %d upload requests", uploadRequests) } last := api.patches[len(api.patches)-1] if last.State() != "completed" { @@ -707,6 +961,9 @@ func TestDownloadShutdownMarksTaskInterrupted(t *testing.T) { if patch.Runtime == nil || patch.Runtime.Message == "" { t.Fatalf("expected interrupted detail message, got %#v", patch.Runtime) } + if eng.resetCalls != 0 { + t.Fatalf("expected interrupted shutdown to preserve resumable runtime data, got %d resets", eng.resetCalls) + } } func TestUploadShutdownMarksTaskInterrupted(t *testing.T) { @@ -739,6 +996,154 @@ func TestUploadShutdownMarksTaskInterrupted(t *testing.T) { } } +func TestSuspendedUploadCleansLocalResultAndForcesRedownload(t *testing.T) { + downloadDir := t.TempDir() + taskDir := filepath.Join(downloadDir, "task-1") + if err := os.MkdirAll(taskDir, 0o755); err != nil { + t.Fatal(err) + } + payloadPath := filepath.Join(taskDir, "payload.bin") + payload := "downloaded payload" + if err := os.WriteFile(payloadPath, []byte(payload), 0o644); err != nil { + t.Fatal(err) + } + payloadSize := int64(len(payload)) + api := &recordingAPI{ + createObjectDraft: client.ObjectDraft{ID: "object-1", Name: "payload.bin", Upload: &client.ObjectUploadInstructions{SessionID: "session-1", PartSize: payloadSize, URLs: []string{"http://127.0.0.1:1"}}}, + } + w := NewWithAPI(config.Config{}, api) + + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(errTaskSuspended) + w.uploadAndComplete( + ctx, + slog.New(slog.NewTextHandler(io.Discard, nil)), + clientTaskWithUploadToken("task-1", "downloading"), + engine.Result{Path: payloadPath, Name: "payload.bin", Size: payloadSize}, + &client.DownloadTaskRuntime{Phase: "uploading"}, + ) + + if _, ok := findPatchWithStatus(api.patches, "suspended"); ok { + t.Fatalf("expected worker not to overwrite server-owned suspended status, got %#v", api.patches) + } + patch := api.patches[len(api.patches)-1] + if patch.Runtime == nil || patch.Runtime.State != localResultRemovedRuntimeState { + t.Fatalf("expected suspended upload cleanup marker, got %#v", patch.Runtime) + } + if _, err := os.Stat(taskDir); !os.IsNotExist(err) { + t.Fatalf("expected suspended upload to remove local task directory, stat err=%v", err) + } + resumed := withRuntime( + withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), payloadSize, &payloadSize), + patch.Runtime, + ) + if got := nextTaskWorkStage(resumed); got != taskWorkStageDownload { + t.Fatalf("expected resumed cleaned upload to redownload, got stage %v", got) + } +} + +func TestProcessRedownloadsWhenLocalResultWasCleaned(t *testing.T) { + api := &recordingAPI{} + eng := &recordingEngine{downloadErr: errors.New("redownload missing local result")} + w := NewWithAPI(config.Config{}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + total := int64(100) + + w.process(context.Background(), withRuntime( + withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), total, &total), + &client.DownloadTaskRuntime{Phase: "error", State: localResultRemovedRuntimeState}, + )) + + if eng.inspectCalls != 0 { + t.Fatalf("expected cleaned local result to skip runtime upload inspection, got %d inspect calls", eng.inspectCalls) + } + if eng.downloadCalls != 1 { + t.Fatalf("expected cleaned local result to resume via download path, got %d download calls", eng.downloadCalls) + } + failed := lastPatchWithStatus(t, api.patches, "failed") + if failed.ErrorMessage == nil || !strings.Contains(*failed.ErrorMessage, "redownload missing local result") { + t.Fatalf("expected resumed redownload failure to be reported, got %#v", failed.ErrorMessage) + } +} + +func TestPausedAndInterruptedDownloadsPreservePartialFiles(t *testing.T) { + cases := []struct { + name string + cancelCtx func(context.CancelCauseFunc, context.CancelFunc) + wantStatus string + }{ + { + name: "paused", + cancelCtx: func(cancelCause context.CancelCauseFunc, _ context.CancelFunc) { + cancelCause(errTaskPausing) + }, + wantStatus: "paused", + }, + { + name: "interrupted", + cancelCtx: func(_ context.CancelCauseFunc, cancel context.CancelFunc) { + cancel() + }, + wantStatus: "interrupted", + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + api := &recordingAPI{} + downloadDir := t.TempDir() + path := filepath.Join(downloadDir, "task-1", "payload.bin") + ready := make(chan struct{}, 1) + eng := &recordingEngine{ + downloadFunc: func(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) { + if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil { + return engine.Result{}, err + } + if err := os.WriteFile(path, []byte("partial"), 0o644); err != nil { + return engine.Result{}, err + } + ready <- struct{}{} + <-ctx.Done() + return engine.Result{}, ctx.Err() + }, + resetTaskFn: func(context.Context, client.DownloadTask) error { + return os.RemoveAll(filepath.Join(downloadDir, "task-1")) + }, + } + + w := NewWithAPI(config.Config{DownloadDir: downloadDir}, api) + w.logger = slog.New(slog.NewTextHandler(io.Discard, nil)) + w.engine = eng + + ctx, cancel := context.WithCancel(context.Background()) + ctxWithCause, cancelCause := context.WithCancelCause(ctx) + done := make(chan struct{}) + go func() { + w.downloadThenUpload(ctxWithCause, w.logger, clientHTTPTask("task-1", "downloading", "https://example.com/payload.bin", "payload.bin"), nil) + close(done) + }() + + select { + case <-ready: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for partial download") + } + tc.cancelCtx(cancelCause, cancel) + waitForWorkerTestCompletion(t, done) + + lastPatchWithStatus(t, api.patches, tc.wantStatus) + info, err := os.Stat(path) + if err != nil { + t.Fatalf("expected resumable file %s to remain, got %v", path, err) + } + if info.Size() == 0 { + t.Fatalf("expected resumable file %s to keep partial content", path) + } + }) + } +} + func TestUploadETARoundsRemainingSeconds(t *testing.T) { eta := uploadETA(&uploadProgress{uploaded: 25, totalBytes: 100}, 20) @@ -838,6 +1243,14 @@ func TestNextTaskWorkStage(t *testing.T) { task: withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 100, &total), want: taskWorkStageUploadExistingResult, }, + { + name: "assigned with removed local result marker", + task: withRuntime( + withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 100, &total), + &client.DownloadTaskRuntime{Phase: "error", State: localResultRemovedRuntimeState}, + ), + want: taskWorkStageDownload, + }, { name: "assigned partial download", task: withDownloadCheckpoint(clientTaskWithStatus("task-1", "assigned"), 99, &total), @@ -977,6 +1390,55 @@ func TestRetainSeedKeepsDownloadedResult(t *testing.T) { } } +func TestRetainedSeedExpiresWhenLedgerPersistenceFails(t *testing.T) { + stateFile := filepath.Join(t.TempDir(), "state-file") + if err := os.WriteFile(stateFile, []byte("not a directory"), 0o644); err != nil { + t.Fatal(err) + } + cleaned := false + w := NewWithAPI(config.Config{SeedEnabled: true, SeedDuration: time.Hour, StateDir: stateFile}, &recordingAPI{}) + w.engine = &recordingEngine{} + + retained := w.retainSeed( + clientTask("task-1"), + engine.Result{ + Path: filepath.Join(t.TempDir(), "result"), + Size: 123, + Seed: &engine.Seed{ + Engine: "aria2", + ID: "gid", + InfoHash: "infohash", + Path: t.TempDir(), + Snapshot: func(context.Context) (engine.SeedSnapshot, error) { + return engine.SeedSnapshot{}, nil + }, + Cleanup: func(context.Context) error { + cleaned = true + return nil + }, + }, + }, + w.logger, + ) + + if !retained { + t.Fatal("expected seed to remain tracked in memory") + } + if len(w.retainedSeedSnapshot()) != 1 { + t.Fatalf("expected retained seed despite ledger failure, got %d", len(w.retainedSeedSnapshot())) + } + w.retainedSeeds[0].expiresAt = time.Now().Add(-time.Second) + + w.cleanupRetainedSeeds(context.Background()) + + if !cleaned { + t.Fatal("expected in-memory retained seed to expire and clean up") + } + if len(w.retainedSeedSnapshot()) != 0 { + t.Fatalf("expected expired seed to be removed from memory, got %d", len(w.retainedSeedSnapshot())) + } +} + func TestReportRetainedSeedsCleansMissingSeed(t *testing.T) { cleaned := false w := NewWithAPI(config.Config{SeedEnabled: true}, &recordingAPI{}) @@ -1317,6 +1779,15 @@ func writeTempFile(t *testing.T, content string) string { return file.Name() } +func waitForWorkerTestCompletion(t *testing.T, done <-chan struct{}) { + t.Helper() + select { + case <-done: + case <-time.After(5 * time.Second): + t.Fatal("timed out waiting for worker completion") + } +} + func clientTask(id string) client.DownloadTask { return client.DownloadTask{ ID: id, @@ -1378,7 +1849,9 @@ func findPatchWithStatus(patches []client.TaskPatch, status string) (client.Task type recordingEngine struct { downloadResult engine.Result downloadErr error + downloadFunc func(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error) resetErr error + resetTaskFn func(context.Context, client.DownloadTask) error taskSnapshot engine.TaskSnapshot inspectErr error inspectPanic any @@ -1425,13 +1898,19 @@ func (e *recordingEngine) ListSeeds(context.Context) ([]engine.Seed, error) { return e.listSeeds, nil } -func (e *recordingEngine) ResetTask(context.Context, client.DownloadTask) error { +func (e *recordingEngine) ResetTask(ctx context.Context, task client.DownloadTask) error { e.resetCalls++ + if e.resetTaskFn != nil { + return e.resetTaskFn(ctx, task) + } return e.resetErr } -func (e *recordingEngine) Download(context.Context, client.DownloadTask, engine.Progress) (engine.Result, error) { +func (e *recordingEngine) Download(ctx context.Context, task client.DownloadTask, progress engine.Progress) (engine.Result, error) { e.downloadCalls++ + if e.downloadFunc != nil { + return e.downloadFunc(ctx, task, progress) + } return e.downloadResult, e.downloadErr } @@ -1439,6 +1918,7 @@ type recordingAPI struct { patches []client.TaskPatch patchedIDs []string seedingTasks []client.DownloadTask + controlTasks []client.DownloadTask assignedTasks []client.DownloadTask suspendDownloading bool createFolderErr error @@ -1451,7 +1931,7 @@ func (a *recordingAPI) Heartbeat(context.Context, client.Heartbeat) error { } func (a *recordingAPI) AssignedControlTasks(context.Context) ([]client.DownloadTask, error) { - return nil, nil + return a.controlTasks, nil } func (a *recordingAPI) AssignedTasks(context.Context) ([]client.DownloadTask, error) { @@ -1469,8 +1949,32 @@ func (a *recordingAPI) UpdateTask(_ context.Context, id string, patch client.Tas if a.suspendDownloading && state == "downloading" { state = "suspended" } + recordedPatch := patch + if state != patch.State() { + recordedPatch.Status = state + } + applyRecordedTaskPatch(a.controlTasks, id, recordedPatch) + applyRecordedTaskPatch(a.assignedTasks, id, recordedPatch) task := clientTaskWithStatus(id, state) - task.Status.Runtime = patch.Runtime + task = applyTaskPatch(task, recordedPatch) + return task, nil +} + +func applyRecordedTaskPatch(tasks []client.DownloadTask, id string, patch client.TaskPatch) { + for i := range tasks { + if tasks[i].ID == id { + tasks[i] = applyTaskPatch(tasks[i], patch) + } + } +} + +func applyTaskPatch(task client.DownloadTask, patch client.TaskPatch) client.DownloadTask { + if patch.State() != "" { + task.Status.State = patch.State() + } + if patch.Runtime != nil { + task.Status.Runtime = patch.Runtime + } if patch.Progress != nil { if patch.Progress.Download != nil { task.Status.Progress.Download = *patch.Progress.Download @@ -1479,7 +1983,7 @@ func (a *recordingAPI) UpdateTask(_ context.Context, id string, patch client.Tas task.Status.Progress.Upload = *patch.Progress.Upload } } - return task, nil + return task } func (a *recordingAPI) CreateFolder(context.Context, string, string, string) (client.ObjectDraft, error) { diff --git a/e2e/helpers.ts b/e2e/helpers.ts index f7e1d4cf..5da5d2e9 100644 --- a/e2e/helpers.ts +++ b/e2e/helpers.ts @@ -117,7 +117,7 @@ export async function pairAndApprove(page: Page): Promise { await expect .poll( async () => { - const result = await getJson(page, `/api/site/licensing/pairings/${pairing.code}`) + const result = await getPairingStatusForPoll(page, pairing.code) if (result.status === 'approved') approved = result return result.status }, @@ -138,6 +138,16 @@ export async function pairAndApprove(page: Page): Promise { return approved } +async function getPairingStatusForPoll(page: Page, code: string): Promise { + const url = `/api/site/licensing/pairings/${code}` + try { + return await getJson(page, url) + } catch (error) { + if (isTransientInternalBrowserJsonError(error, 'GET', url)) return { status: 'pending' } + throw error + } +} + export async function approvePairingInCloud(pairing: PairingInfo) { const email = process.env.E2E_CLOUD_BUSINESS_EMAIL ?? process.env.E2E_CLOUD_PRO_EMAIL const password = process.env.E2E_CLOUD_BUSINESS_PASSWORD ?? process.env.E2E_CLOUD_PRO_PASSWORD @@ -271,3 +281,8 @@ function isTransientBrowserJsonError(error: unknown) { error.message.includes('Load failed') ) } + +function isTransientInternalBrowserJsonError(error: unknown, method: 'GET' | 'POST', url: string) { + if (!(error instanceof Error)) return false + return error.message.includes(`${method} ${url} failed with 500:`) && error.message.includes('"status":"INTERNAL"') +}