fix(downloader): supervise managed engines and make the CLI go-installable (#461)

Three related changes hardening the remote downloader, bundled because the
module-path rename touches every file — splitting would leave a mid-history
commit that doesn't build.

Engine supervision: a managed engine subprocess (aria2c / qbittorrent-nox)
that exits unexpectedly now crashes the worker. watchEngineProcess waits on
the child, logs the exit at error level, and cancels the run context so
`downloader up` returns non-zero; the container (restart: unless-stopped)
then restarts the whole stack. Previously the exit error was discarded and
the worker kept heartbeating as healthy while every task failed against the
dead RPC. A deliberate shutdown kill is told apart from a crash via a
stopping flag.

aria2 GID recovery: waitAria2 re-discovers the live GID via findTask when
aria2 reports 'GID ... is not found' mid-download (e.g. after an aria2
restart) instead of failing the task. Adds isAria2GIDNotFound, which matches
tellStatus's message format that isAria2DownloadNotFound missed.

CLI install: move the entrypoint to the module root (cmd/main.go) and rename
the module github.com/saltbo/zpan/cmd -> github.com/saltbo/zpan so
`go install .` from cmd/ yields a zpan binary. Internal imports shorten to
github.com/saltbo/zpan/internal/...; the Dockerfile builds the module root.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Dockerfile

@@ -27,7 +27,7 @@ WORKDIR /app/cmd
 COPY cmd/go.mod cmd/go.sum ./
 RUN go mod download
 COPY cmd ./
-RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/zpan ./zpan
+RUN CGO_ENABLED=0 go build -trimpath -ldflags="-s -w" -o /out/zpan .
 
 FROM debian:bookworm-slim AS geoip-db
 ARG GEOIP_DB_MONTH=2026-06

cmd/go.mod

@@ -1,4 +1,4 @@
-module github.com/saltbo/zpan/cmd
+module github.com/saltbo/zpan
 
 go 1.25.0
 

cmd/internal/client/client.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/openapi"
+	"github.com/saltbo/zpan/internal/openapi"
 )
 
 type Client struct {

cmd/internal/engine/aria2.go

@@ -19,7 +19,7 @@ import (
 
 	"github.com/Braurbeki/arigo"
 	"github.com/cenkalti/rpc2"
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 type Aria2 struct {
@@ -74,7 +74,6 @@ func (a Aria2) Start(ctx context.Context) (*exec.Cmd, error) {
 	if err := cmd.Start(); err != nil {
 		return nil, err
 	}
-	go func() { _ = cmd.Wait() }()
 	return cmd, nil
 }
 
@@ -193,6 +192,20 @@ func isAria2DownloadNotFound(err error) bool {
 	return strings.Contains(msg, "download") && strings.Contains(msg, "not found")
 }
 
+// isAria2GIDNotFound reports whether a status lookup failed because aria2 has no
+// record of the GID — the message aria2 returns from tellStatus is
+// "GID <gid> is not found", which lacks the word "download" that
+// isAria2DownloadNotFound looks for. This happens when aria2 restarts mid-task
+// and re-creates the download under a fresh GID.
+func isAria2GIDNotFound(err error) bool {
+	if err == nil {
+		return false
+	}
+	msg := strings.ToLower(err.Error())
+	return strings.Contains(msg, "not found") &&
+		(strings.Contains(msg, "gid") || strings.Contains(msg, "download"))
+}
+
 func (a Aria2) SaveSession(ctx context.Context) error {
 	client, err := a.client(ctx)
 	if err != nil {
@@ -412,7 +425,7 @@ func (a Aria2) waitResult(ctx context.Context, aria **arigo.Client, task client.
 		initialProgress = func(downloaded int64, total *int64, bps int64, detail *client.DownloadTaskRuntime) error { return nil }
 	}
 	primaryGID := gid
-	status, err := a.waitAria2(ctx, aria, primaryGID, initialProgress)
+	status, err := a.waitAria2(ctx, aria, task, primaryGID, initialProgress)
 	if err != nil {
 		_ = (*aria).Remove(primaryGID)
 		return Result{}, fmt.Errorf("wait primary gid %s: %w", primaryGID, err)
@@ -420,7 +433,7 @@ func (a Aria2) waitResult(ctx context.Context, aria **arigo.Client, task client.
 	resultGID := status.GID
 	if len(status.FollowedBy) > 0 {
 		childGID := status.FollowedBy[0]
-		status, err = a.waitAria2(ctx, aria, childGID, progress)
+		status, err = a.waitAria2(ctx, aria, task, childGID, progress)
 		if err != nil {
 			_ = (*aria).Remove(childGID)
 			return Result{}, fmt.Errorf("wait followed gid %s: %w", childGID, err)
@@ -741,7 +754,7 @@ func (a Aria2) cleanupSeed(gid string, localPath string) func(context.Context) e
 	}
 }
 
-func (a Aria2) waitAria2(ctx context.Context, aria **arigo.Client, gid string, progress Progress) (arigo.Status, error) {
+func (a Aria2) waitAria2(ctx context.Context, aria **arigo.Client, task client.DownloadTask, gid string, progress Progress) (arigo.Status, error) {
 	ticker := time.NewTicker(time.Second)
 	defer ticker.Stop()
 	for {
@@ -757,6 +770,17 @@ func (a Aria2) waitAria2(ctx context.Context, aria **arigo.Client, gid string, p
 					}
 					continue
 				}
+				if isAria2GIDNotFound(err) {
+					recovered, ok, findErr := a.findTask(ctx, aria, task)
+					if findErr != nil {
+						return arigo.Status{}, findErr
+					}
+					if !ok || recovered.GID == gid {
+						return arigo.Status{}, err
+					}
+					gid = recovered.GID
+					continue
+				}
 				return arigo.Status{}, err
 			}
 			total := int64(status.TotalLength)

cmd/internal/engine/engine.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 type Result struct {

cmd/internal/engine/engine_test.go

@@ -14,7 +14,7 @@ import (
 	"github.com/Braurbeki/arigo"
 	qbittorrent "github.com/autobrr/go-qbittorrent"
 	"github.com/cenkalti/rpc2"
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 func downloadTask(id, sourceType, sourceURI string) client.DownloadTask {
@@ -226,6 +226,18 @@ func TestIsAria2DownloadNotFound(t *testing.T) {
 	}
 }
 
+func TestIsAria2GIDNotFound(t *testing.T) {
+	if !isAria2GIDNotFound(errors.New("GID 8bddd19e07ad6dc3 is not found")) {
+		t.Fatal("expected tellStatus GID-not-found to trigger re-discovery")
+	}
+	if !isAria2GIDNotFound(errors.New("Active Download not found for GID#b384ccaa7eae88da")) {
+		t.Fatal("expected aria2 active download not found to trigger re-discovery")
+	}
+	if isAria2GIDNotFound(errors.New("aria2 download ended with status error")) {
+		t.Fatal("expected ordinary aria2 download errors to stay visible")
+	}
+}
+
 func TestHTTPDownloadResumesExistingFile(t *testing.T) {
 	var rangeHeader string
 	server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {

cmd/internal/engine/geoip.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 
 	"github.com/oschwald/geoip2-golang"
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 var geoIPCodePattern = regexp.MustCompile(`^[A-Z0-9-]+$`)

cmd/internal/engine/http.go

@@ -12,7 +12,7 @@ import (
 	"strconv"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 type HTTP struct {

cmd/internal/engine/live_download_test.go

@@ -18,7 +18,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 func liveTask(id, sourceType, sourceURI, name string) client.DownloadTask {

cmd/internal/engine/qbittorrent.go

@@ -13,7 +13,7 @@ import (
 	"time"
 
 	qbittorrent "github.com/autobrr/go-qbittorrent"
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 type QBittorrent struct {
@@ -49,7 +49,6 @@ func (q QBittorrent) Start(ctx context.Context) (*exec.Cmd, error) {
 	if err := cmd.Start(); err != nil {
 		return nil, err
 	}
-	go func() { _ = cmd.Wait() }()
 	return cmd, nil
 }
 

cmd/internal/worker/api.go

@@ -5,7 +5,7 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
+	"github.com/saltbo/zpan/internal/client"
 )
 
 const apiRetryAttempts = 3

cmd/internal/worker/api_test.go

@@ -5,7 +5,7 @@ import (
 	"errors"
 	"testing"
 
-	"github.com/saltbo/zpan/cmd/internal/config"
+	"github.com/saltbo/zpan/internal/config"
 )
 
 func TestCallAPIRetriesTransientErrors(t *testing.T) {

cmd/internal/worker/engines.go

@@ -3,11 +3,12 @@ package worker
 import (
 	"context"
 	"fmt"
+	"os/exec"
 	"strings"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/config"
-	"github.com/saltbo/zpan/cmd/internal/engine"
+	"github.com/saltbo/zpan/internal/config"
+	"github.com/saltbo/zpan/internal/engine"
 )
 
 func (w *Worker) resolveEngine(ctx context.Context) error {
@@ -86,12 +87,35 @@ func (w *Worker) startEngine(ctx context.Context, downloader engine.Engine) erro
 		if cmd.Process != nil {
 			_ = cmd.Process.Kill()
 		}
+		_ = cmd.Wait()
 		return err
 	}
+	go w.watchEngineProcess(downloader.Name(), cmd)
 	return nil
 }
 
+// watchEngineProcess waits on a managed engine subprocess for the worker's
+// lifetime. The engine is expected to outlive every task, so any exit we did
+// not initiate is fatal: log it and cancel the run context, which makes Run
+// return errEngineExited and the process exit non-zero so the supervisor
+// restarts the whole downloader.
+func (w *Worker) watchEngineProcess(name string, cmd *exec.Cmd) {
+	err := cmd.Wait()
+	if w.isStopping() {
+		return
+	}
+	pid := 0
+	if cmd.Process != nil {
+		pid = cmd.Process.Pid
+	}
+	w.logger.Error("managed downloader engine exited unexpectedly", "engine", name, "pid", pid, "error", err)
+	if w.cancelRun != nil {
+		w.cancelRun(fmt.Errorf("%w: %s (pid %d): %v", errEngineExited, name, pid, err))
+	}
+}
+
 func (w *Worker) stopStartedEngines() {
+	w.markStopping()
 	if len(w.started) > 0 {
 		if saver, ok := w.engine.(engine.SessionSaver); ok {
 			ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

cmd/internal/worker/seeds.go

@@ -8,8 +8,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
-	"github.com/saltbo/zpan/cmd/internal/engine"
+	"github.com/saltbo/zpan/internal/client"
+	"github.com/saltbo/zpan/internal/engine"
 )
 
 const retainedSeedReportInterval = 5 * time.Second

cmd/internal/worker/uploader.go

@@ -11,8 +11,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
-	"github.com/saltbo/zpan/cmd/internal/engine"
+	"github.com/saltbo/zpan/internal/client"
+	"github.com/saltbo/zpan/internal/engine"
 )
 
 type uploadProgress struct {

cmd/internal/worker/worker.go

@@ -15,10 +15,10 @@ import (
 	"sync"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
-	"github.com/saltbo/zpan/cmd/internal/config"
-	"github.com/saltbo/zpan/cmd/internal/engine"
-	"github.com/saltbo/zpan/cmd/internal/host"
+	"github.com/saltbo/zpan/internal/client"
+	"github.com/saltbo/zpan/internal/config"
+	"github.com/saltbo/zpan/internal/engine"
+	"github.com/saltbo/zpan/internal/host"
 )
 
 const Version = "0.1.0"
@@ -27,6 +27,7 @@ const maxTaskErrorMessageLength = 1000
 var errBillingPaused = errors.New("billing paused")
 var errTaskPausing = errors.New("task pausing")
 var errTaskCanceling = errors.New("task canceling")
+var errEngineExited = errors.New("managed downloader engine exited")
 
 type taskWorkStage int
 
@@ -46,6 +47,8 @@ type Worker struct {
 	retainedSeeds []retainedSeed
 	attempts      map[string]int
 	started       []*exec.Cmd
+	cancelRun     context.CancelCauseFunc
+	stopping      bool
 	wg            sync.WaitGroup
 	mu            sync.Mutex
 }
@@ -114,12 +117,20 @@ func (w *Worker) Run(ctx context.Context) error {
 		w.logger.Info("geoip database loaded", "path", w.cfg.GeoIPDBPath)
 		defer w.geoIP.Close()
 	}
-	if err := w.resolveEngine(ctx); err != nil {
+	// runCtx is cancelled either by the parent ctx (signal-driven shutdown) or
+	// by watchEngineProcess when a managed engine subprocess dies. The latter
+	// surfaces errEngineExited as the cancel cause so Run returns a non-nil
+	// error and the process exits non-zero for the supervisor to restart.
+	runCtx, cancelRun := context.WithCancelCause(ctx)
+	defer cancelRun(nil)
+	w.cancelRun = cancelRun
+
+	if err := w.resolveEngine(runCtx); err != nil {
 		return err
 	}
 	defer w.stopStartedEngines()
 
-	checkCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
+	checkCtx, cancel := context.WithTimeout(runCtx, 10*time.Second)
 	defer cancel()
 	w.logger.Info("checking downloader engine", "engine", w.cfg.Engine)
 	if err := w.engine.Check(checkCtx); err != nil {
@@ -128,7 +139,7 @@ func (w *Worker) Run(ctx context.Context) error {
 	}
 	w.logger.Info("downloader engine check passed", "engine", w.cfg.Engine)
 	w.logger.Info("downloader started", "engine", w.cfg.Engine)
-	w.restoreRetainedSeeds(ctx)
+	w.restoreRetainedSeeds(runCtx)
 	ticker := time.NewTicker(w.cfg.PollInterval)
 	defer ticker.Stop()
 	seedCleanupTicker := time.NewTicker(time.Minute)
@@ -136,25 +147,30 @@ func (w *Worker) Run(ctx context.Context) error {
 	seedReportTicker := time.NewTicker(retainedSeedReportInterval)
 	defer seedReportTicker.Stop()
 
-	if err := w.tick(ctx); err != nil {
+	if err := w.tick(runCtx); err != nil {
 		w.logger.Error("downloader tick failed", "error", err)
 	}
 	for {
 		select {
-		case <-ctx.Done():
-			w.logger.Info("downloader stopped", "reason", ctx.Err())
-			w.reportRetainedSeedsStopped(context.WithoutCancel(ctx))
+		case <-runCtx.Done():
+			cause := context.Cause(runCtx)
+			w.reportRetainedSeedsStopped(context.WithoutCancel(runCtx))
 			w.waitForTasks()
+			if errors.Is(cause, errEngineExited) {
+				// watchEngineProcess already logged the exit at error level.
+				return cause
+			}
+			w.logger.Info("downloader stopped", "reason", cause)
 			return nil
 		case <-ticker.C:
-			if err := w.tick(ctx); err != nil {
+			if err := w.tick(runCtx); err != nil {
 				w.logger.Error("downloader tick failed", "error", err)
 			}
 		case <-seedCleanupTicker.C:
-			w.cleanupRetainedSeeds(ctx)
+			w.cleanupRetainedSeeds(runCtx)
 		case <-seedReportTicker.C:
-			w.restoreRetainedSeeds(ctx)
-			w.reportRetainedSeeds(ctx)
+			w.restoreRetainedSeeds(runCtx)
+			w.reportRetainedSeeds(runCtx)
 		}
 	}
 }
@@ -718,6 +734,20 @@ func (w *Worker) finish(taskID string) {
 	}
 }
 
+// markStopping records that the worker is shutting down on purpose so
+// watchEngineProcess can tell a deliberate engine kill from a crash.
+func (w *Worker) markStopping() {
+	w.mu.Lock()
+	w.stopping = true
+	w.mu.Unlock()
+}
+
+func (w *Worker) isStopping() bool {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	return w.stopping
+}
+
 func (w *Worker) waitForTasks() {
 	done := make(chan struct{})
 	go func() {

cmd/internal/worker/worker_test.go

@@ -8,17 +8,68 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"os"
+	"os/exec"
 	"path/filepath"
 	"strconv"
 	"strings"
 	"testing"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
-	"github.com/saltbo/zpan/cmd/internal/config"
-	"github.com/saltbo/zpan/cmd/internal/engine"
+	"github.com/saltbo/zpan/internal/client"
+	"github.com/saltbo/zpan/internal/config"
+	"github.com/saltbo/zpan/internal/engine"
 )
 
+func TestWatchEngineProcessFatalOnUnexpectedExit(t *testing.T) {
+	w := NewWithAPI(config.Config{}, nil)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	runCtx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+	w.cancelRun = cancel
+
+	cmd := exec.Command("sleep", "30")
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start managed process stub: %v", err)
+	}
+	done := make(chan struct{})
+	go func() {
+		w.watchEngineProcess("aria2", cmd)
+		close(done)
+	}()
+	// Simulate the engine dying out from under the worker.
+	_ = cmd.Process.Kill()
+
+	select {
+	case <-runCtx.Done():
+	case <-time.After(5 * time.Second):
+		t.Fatal("expected run context to be cancelled after the engine exits")
+	}
+	if cause := context.Cause(runCtx); !errors.Is(cause, errEngineExited) {
+		t.Fatalf("expected errEngineExited cause, got %v", cause)
+	}
+	<-done
+}
+
+func TestWatchEngineProcessQuietOnDeliberateStop(t *testing.T) {
+	w := NewWithAPI(config.Config{}, nil)
+	w.logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+	runCtx, cancel := context.WithCancelCause(context.Background())
+	defer cancel(nil)
+	w.cancelRun = cancel
+	w.markStopping()
+
+	cmd := exec.Command("sleep", "30")
+	if err := cmd.Start(); err != nil {
+		t.Fatalf("start managed process stub: %v", err)
+	}
+	_ = cmd.Process.Kill()
+	w.watchEngineProcess("aria2", cmd)
+
+	if cause := context.Cause(runCtx); cause != nil {
+		t.Fatalf("expected run context to stay live during a deliberate stop, got %v", cause)
+	}
+}
+
 func TestHeartbeatReportsAggregateTransferSpeeds(t *testing.T) {
 	w := NewWithAPI(config.Config{Engine: "auto", MaxConcurrentTasks: 5}, &recordingAPI{})
 

cmd/main.go

@@ -11,10 +11,10 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/saltbo/zpan/cmd/internal/client"
-	"github.com/saltbo/zpan/cmd/internal/config"
-	"github.com/saltbo/zpan/cmd/internal/host"
-	"github.com/saltbo/zpan/cmd/internal/worker"
+	"github.com/saltbo/zpan/internal/client"
+	"github.com/saltbo/zpan/internal/config"
+	"github.com/saltbo/zpan/internal/host"
+	"github.com/saltbo/zpan/internal/worker"
 	"github.com/spf13/cobra"
 	"github.com/spf13/viper"
 )

cmd/main_test.go

@@ -3,7 +3,7 @@ package main
 import (
 	"testing"
 
-	"github.com/saltbo/zpan/cmd/internal/config"
+	"github.com/saltbo/zpan/internal/config"
 )
 
 func TestRegistrationHeartbeatNormalizesAutoEngine(t *testing.T) {