From 2ed46ea9688a807449512ed8e0e62227ff4f7e4f Mon Sep 17 00:00:00 2001 From: saltbo Date: Fri, 5 Jun 2026 21:14:02 -0400 Subject: [PATCH] fix(downloader): exit cleanly on shutdown --- downloader/internal/worker/worker.go | 28 +++++++++++++++++++++++++--- 1 file changed, 25 insertions(+), 3 deletions(-) diff --git a/downloader/internal/worker/worker.go b/downloader/internal/worker/worker.go index 46540f6d..47efb30e 100644 --- a/downloader/internal/worker/worker.go +++ b/downloader/internal/worker/worker.go @@ -41,6 +41,7 @@ type Worker struct { running map[string]context.CancelCauseFunc retainedSeeds []retainedSeed started []*exec.Cmd + wg sync.WaitGroup mu sync.Mutex } @@ -120,7 +121,8 @@ func (w *Worker) Run(ctx context.Context) error { select { case <-ctx.Done(): w.logger.Info("downloader stopped", "reason", ctx.Err()) - return ctx.Err() + w.waitForTasks() + return nil case <-ticker.C: if err := w.tick(ctx); err != nil { w.logger.Error("downloader tick failed", "error", err) @@ -499,13 +501,33 @@ func (w *Worker) startTask(ctx context.Context, taskID string) (context.Context, } taskCtx, cancel := context.WithCancelCause(ctx) w.running[taskID] = cancel + w.wg.Add(1) return taskCtx, true } func (w *Worker) finish(taskID string) { w.mu.Lock() - defer w.mu.Unlock() - delete(w.running, taskID) + _, exists := w.running[taskID] + if exists { + delete(w.running, taskID) + } + w.mu.Unlock() + if exists { + w.wg.Done() + } +} + +func (w *Worker) waitForTasks() { + done := make(chan struct{}) + go func() { + w.wg.Wait() + close(done) + }() + select { + case <-done: + case <-time.After(10 * time.Second): + w.logger.Warn("timed out waiting for running tasks to stop", "running", w.currentTasks()) + } } func (w *Worker) cancelRunning(task client.DownloadTask) bool {