feat(agent): job runner, compose override executor and swarm executor (#7)

go.mod

@@ -5,23 +5,22 @@ go 1.26.3
 require (
 	github.com/coder/websocket v1.8.15
 	github.com/coreos/go-oidc/v3 v3.21.0
+	github.com/distribution/reference v0.6.0
 	github.com/docker/docker v28.5.2+incompatible
 	github.com/jackc/pgx/v5 v5.10.0
-	github.com/prometheus/client_golang v1.23.2
+	github.com/opencontainers/go-digest v1.0.0
-	github.com/prometheus/client_model v0.6.2
 	github.com/stretchr/testify v1.11.1
 	golang.org/x/oauth2 v0.36.0
+	gopkg.in/yaml.v3 v3.0.1
 )
 
 require (
 	github.com/Microsoft/go-winio v0.6.2 // indirect
-	github.com/beorn7/perks v1.0.1 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
 	github.com/containerd/errdefs v1.0.0 // indirect
 	github.com/containerd/errdefs/pkg v0.3.0 // indirect
 	github.com/containerd/log v0.1.0 // indirect
 	github.com/davecgh/go-spew v1.1.1 // indirect
-	github.com/distribution/reference v0.6.0 // indirect
 	github.com/docker/go-connections v0.7.0 // indirect
 	github.com/docker/go-units v0.5.0 // indirect
 	github.com/felixge/httpsnoop v1.0.4 // indirect
@@ -35,25 +34,18 @@ require (
 	github.com/moby/sys/atomicwriter v0.1.0 // indirect
 	github.com/moby/term v0.5.2 // indirect
 	github.com/morikuni/aec v1.1.0 // indirect
-	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
-	github.com/opencontainers/go-digest v1.0.0 // indirect
 	github.com/opencontainers/image-spec v1.1.1 // indirect
 	github.com/pkg/errors v0.9.1 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
-	github.com/prometheus/common v0.66.1 // indirect
-	github.com/prometheus/procfs v0.16.1 // indirect
 	go.opentelemetry.io/auto/sdk v1.2.1 // indirect
 	go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.68.0 // indirect
 	go.opentelemetry.io/otel v1.43.0 // indirect
 	go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.43.0 // indirect
 	go.opentelemetry.io/otel/metric v1.43.0 // indirect
 	go.opentelemetry.io/otel/trace v1.43.0 // indirect
-	go.yaml.in/yaml/v2 v2.4.2 // indirect
 	golang.org/x/sync v0.20.0 // indirect
 	golang.org/x/sys v0.42.0 // indirect
 	golang.org/x/text v0.35.0 // indirect
 	golang.org/x/time v0.15.0 // indirect
-	google.golang.org/protobuf v1.36.11 // indirect
-	gopkg.in/yaml.v3 v3.0.1 // indirect
 	gotest.tools/v3 v3.5.2 // indirect
 )

go.sum

@@ -2,8 +2,6 @@ github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c h1:udKWzYgxTojEK
 github.com/Azure/go-ansiterm v0.0.0-20250102033503-faa5f7b0171c/go.mod h1:xomTg63KZ2rFqZQzSB4Vz2SUXa1BpHTVz9L5PTmPC4E=
 github.com/Microsoft/go-winio v0.6.2 h1:F2VQgta7ecxGYO8k3ZZz3RS8fVIXVxONVUPlNERoyfY=
 github.com/Microsoft/go-winio v0.6.2/go.mod h1:yd8OoFMLzJbo9gZq8j5qaps8bJ9aShtEA8Ipt1oGCvU=
-github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM=
-github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw=
 github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
 github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
 github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs=
@@ -52,14 +50,10 @@ github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
 github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
 github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
 github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
-github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo=
-github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ=
 github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
 github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc=
-github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw=
 github.com/moby/docker-image-spec v1.3.1 h1:jMKff3w6PgbfSa69GfNg+zN/XLhfXJGnEx3Nl2EsFP0=
 github.com/moby/docker-image-spec v1.3.1/go.mod h1:eKmb5VW8vQEh/BAr2yvVNvuiJuY6UIocYsFu/DxxRpo=
 github.com/moby/sys/atomicwriter v0.1.0 h1:kw5D/EqkBwsBFi0ss9v1VG3wIkVhzGvLklJ+w3A14Sw=
@@ -70,8 +64,6 @@ github.com/moby/term v0.5.2 h1:6qk3FJAFDs6i/q3W/pQ97SX192qKfZgGjCQqfCJkgzQ=
 github.com/moby/term v0.5.2/go.mod h1:d3djjFCrjnB+fl8NJux+EJzu0msscUP+f8it8hPkFLc=
 github.com/morikuni/aec v1.1.0 h1:vBBl0pUnvi/Je71dsRrhMBtreIqNMYErSAbEeb8jrXQ=
 github.com/morikuni/aec v1.1.0/go.mod h1:xDRgiq/iw5l+zkao76YTKzKttOp2cwPEne25HDkJnBw=
-github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA=
-github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ=
 github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U=
 github.com/opencontainers/go-digest v1.0.0/go.mod h1:0JzlMkj0TRzQZfJkVvzbP0HBR3IKzErnv2BNG4W4MAM=
 github.com/opencontainers/image-spec v1.1.1 h1:y0fUlFfIZhPF1W537XOLg0/fcx6zcHCJwooC2xJA040=
@@ -80,14 +72,6 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o=
-github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg=
-github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk=
-github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE=
-github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs=
-github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA=
-github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg=
-github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is=
 github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
 github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
 github.com/sirupsen/logrus v1.9.3 h1:dueUQJ1C2q9oE3F7wvmSGAaVtTmUizReu6fjN8uqzbQ=
@@ -117,10 +101,6 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g=
 go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk=
-go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
-go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
-go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI=
-go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU=
 golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0=
 golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw=
 golang.org/x/oauth2 v0.36.0 h1:peZ/1z27fi9hUOFCAZaHyrpWG5lwe0RJEEEeH0ThlIs=

internal/agent/updater/compose.go

@@ -3,52 +3,129 @@ package updater
 import (
 	"context"
 	"fmt"
+	"io"
 	"os/exec"
 	"strings"
+	"time"
 
-	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/filters"
+
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
-// ComposeExecutor invokes `docker compose` as a subprocess.
+const (
-type ComposeExecutor struct{}
+	labelComposeProject = "com.docker.compose.project"
+	labelComposeService = "com.docker.compose.service"
+	labelWorkingDir     = "com.docker.compose.project.working_dir"
+	labelConfigFiles    = "com.docker.compose.project.config_files"
+)
 
-// NewComposeExecutor returns an executor that shells out to docker compose.
+// errorOutputTail is how much of a failed command's output goes into the
-func NewComposeExecutor() *ComposeExecutor {
+// job's error message; the full output is in the job log.
-	return &ComposeExecutor{}
+const errorOutputTail = 2 << 10
+
+// tailWriter adapts a tailBuffer to io.Writer.
+type tailWriter struct{ b *tailBuffer }
+
+func (w tailWriter) Write(p []byte) (int, error) {
+	w.b.Write(p)
+	return len(p), nil
 }
 
-// Execute runs `docker compose -f <file>... -p <project> pull <service>`
+// CommandRunner runs "docker <args>" in dir with stdout and stderr to out.
-// followed by `... up -d <service>`. Working directory is set to
+type CommandRunner func(ctx context.Context, dir string, out io.Writer, args ...string) error
-// job.WorkingDir so any relative paths in the compose file resolve correctly.
+
-func (e *ComposeExecutor) Execute(ctx context.Context, job discovery.Job) error {
+// ExecDocker is the real CommandRunner. When ctx ends the process is killed
-	if job.Refused {
+// and, after a grace period, its output pipes are closed.
-		return fmt.Errorf("refused: %s", job.RefusedReason)
+func ExecDocker(ctx context.Context, dir string, out io.Writer, args ...string) error {
-	}
+	cmd := exec.CommandContext(ctx, "docker", args...)
-	if err := e.run(ctx, job, "pull"); err != nil {
+	cmd.Dir = dir
-		return fmt.Errorf("pull: %w", err)
+	cmd.Stdout = out
-	}
+	cmd.Stderr = out
-	if err := e.run(ctx, job, "up", "-d"); err != nil {
+	cmd.WaitDelay = 10 * time.Second
-		return fmt.Errorf("up: %w", err)
+	return cmd.Run()
-	}
-	return nil
 }
 
-func (e *ComposeExecutor) run(ctx context.Context, job discovery.Job, args ...string) error {
+// ComposeDocker is the subset of the Docker SDK ComposeExecutor needs.
-	cliArgs := []string{"compose"}
+type ComposeDocker interface {
-	for _, f := range job.ConfigFiles {
+	SelfInspector
-		cliArgs = append(cliArgs, "-f", f)
+	ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error)
-	}
+}
-	cliArgs = append(cliArgs, "-p", job.Project)
-	cliArgs = append(cliArgs, args...)
-	cliArgs = append(cliArgs, job.Service)
 
-	cmd := exec.CommandContext(ctx, "docker", cliArgs...)
+// ComposeExecutor updates a Compose service by writing the agent-owned
-	cmd.Dir = job.WorkingDir
+// override file and running `docker compose pull` and `up -d`.
+type ComposeExecutor struct {
+	cli     ComposeDocker
+	resolve *Resolver
+	selfID  string
+	run     CommandRunner
+}
 
-	out, err := cmd.CombinedOutput()
+// NewComposeExecutor returns an executor for Compose jobs. selfID names the
+// agent's own container, whose mounts decide which directories may be
+// touched; run is ExecDocker outside tests.
+func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run CommandRunner) *ComposeExecutor {
+	return &ComposeExecutor{cli: cli, resolve: resolve, selfID: selfID, run: run}
+}
+
+// Execute pins job.Ref to its current digest, writes it to the stack's
+// override file and recreates the service. Working directory and config
+// files come from the service's container labels on this machine, never from
+// the job. It returns the pinned ref.
+func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
+	project, service := job.ComposeProject, job.ComposeService
+	cts, err := e.cli.ContainerList(ctx, container.ListOptions{
+		All: true,
+		Filters: filters.NewArgs(
+			filters.Arg("label", labelComposeProject+"="+project),
+			filters.Arg("label", labelComposeService+"="+service),
+		),
+	})
 	if err != nil {
-		return fmt.Errorf("docker %s: %w (output: %s)",
+		return "", fmt.Errorf("list containers: %w", err)
-			strings.Join(args, " "), err, strings.TrimSpace(string(out)))
 	}
-	return nil
+	if len(cts) == 0 {
+		return "", fmt.Errorf("no container for compose service %s/%s on this machine", project, service)
+	}
+	workingDir := cts[0].Labels[labelWorkingDir]
+	configFiles := cts[0].Labels[labelConfigFiles]
+
+	roots, err := MountRoots(ctx, e.cli, e.selfID)
+	if err != nil {
+		return "", err
+	}
+	files, dir, err := composeFiles(roots, configFiles)
+	if err != nil {
+		return "", err
+	}
+	if workingDir == "" {
+		workingDir = dir
+	} else if !insideAny(roots, workingDir) {
+		return "", notMounted(workingDir)
+	}
+
+	pinned, err := e.resolve.Resolve(ctx, job.Ref)
+	if err != nil {
+		return "", err
+	}
+	fmt.Fprintf(out, "resolved %s\n", pinned)
+
+	if _, err := WriteOverride(dir, service, pinned); err != nil {
+		return "", err
+	}
+
+	base := []string{"compose", "-p", project}
+	for _, f := range files {
+		base = append(base, "-f", f)
+	}
+	for _, step := range [][]string{{"pull", service}, {"up", "-d", service}} {
+		args := append(append([]string(nil), base...), step...)
+		fmt.Fprintf(out, "$ docker %s\n", strings.Join(args, " "))
+		tail := &tailBuffer{max: errorOutputTail}
+		if err := e.run(ctx, workingDir, io.MultiWriter(out, tailWriter{tail}), args...); err != nil {
+			return "", fmt.Errorf("docker compose %s: %w (output: %s)", step[0], err, strings.TrimSpace(tail.String()))
+		}
+	}
+	return pinned, nil
 }

internal/agent/updater/compose_test.go

@@ -0,0 +1,132 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"io"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/mount"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type fakeCompose struct {
+	fakeSelf
+	containers []container.Summary
+}
+
+func (f fakeCompose) ContainerList(context.Context, container.ListOptions) ([]container.Summary, error) {
+	return f.containers, nil
+}
+
+type ranCmd struct {
+	dir  string
+	args []string
+}
+
+type recorder struct {
+	cmds       []ranCmd
+	failOn     string // second arg after -f list, e.g. "pull"
+	failOutput string
+	onFirst    func()
+}
+
+func (r *recorder) run(_ context.Context, dir string, out io.Writer, args ...string) error {
+	if len(r.cmds) == 0 && r.onFirst != nil {
+		r.onFirst()
+	}
+	r.cmds = append(r.cmds, ranCmd{dir, args})
+	if r.failOn != "" && strings.Contains(strings.Join(args, " "), " "+r.failOn+" ") {
+		io.WriteString(out, r.failOutput)
+		return errors.New("exit status 1")
+	}
+	return nil
+}
+
+func TestComposeExecutor(t *testing.T) {
+	const d = "sha256:d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1"
+	root, _ := filepath.EvalSymlinks(t.TempDir())
+	stack := filepath.Join(root, "slusk")
+	require.NoError(t, os.MkdirAll(stack, 0o755))
+	job := protocol.JobStart{Kind: protocol.KindCompose, ComposeProject: "slusk", ComposeService: "web", Ref: "gitea.shcizo.se/shcizo/slusk:1.4"}
+	pinned := job.Ref + "@" + d
+
+	ct := container.Summary{Labels: map[string]string{
+		labelWorkingDir:  stack,
+		labelConfigFiles: filepath.Join(stack, "compose.yaml"),
+	}}
+	rw := fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeBind, Source: root, Destination: root, RW: true}}}
+	ro := fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeBind, Source: root, Destination: root, RW: false}}}
+	wantFiles := []string{"-f", filepath.Join(stack, "compose.yaml"), "-f", filepath.Join(stack, OverrideName)}
+	cmd := func(rest ...string) []string {
+		return append(append([]string{"compose", "-p", "slusk"}, wantFiles...), rest...)
+	}
+	overridePath := filepath.Join(stack, OverrideName)
+	cleanOverride := func() { os.Remove(overridePath) }
+
+	setup := func(cli ComposeDocker, dist *fakeDist, rec *recorder) *ComposeExecutor {
+		return NewComposeExecutor(cli, NewResolver(dist, nil), "me", rec.run)
+	}
+
+	t.Run("happy path", func(t *testing.T) {
+		t.Cleanup(cleanOverride)
+		rec := &recorder{onFirst: func() {
+			_, err := os.Stat(overridePath)
+			assert.NoError(t, err, "override must exist before the first command")
+		}}
+		var out strings.Builder
+		got, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, &out)
+		require.NoError(t, err)
+		assert.Equal(t, pinned, got)
+		require.Len(t, rec.cmds, 2)
+		assert.Equal(t, ranCmd{stack, cmd("pull", "web")}, rec.cmds[0])
+		assert.Equal(t, ranCmd{stack, cmd("up", "-d", "web")}, rec.cmds[1])
+		_, doc := readOverride(t, stack)
+		assert.Equal(t, pinned, serviceImage(doc, "web"))
+		assert.Contains(t, out.String(), "resolved "+pinned)
+	})
+	t.Run("no matching container", func(t *testing.T) {
+		rec := &recorder{}
+		_, err := setup(fakeCompose{rw, nil}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard)
+		assert.ErrorContains(t, err, "no container for compose service slusk/web on this machine")
+		assert.Empty(t, rec.cmds)
+	})
+	t.Run("resolve fails", func(t *testing.T) {
+		t.Cleanup(cleanOverride)
+		rec := &recorder{}
+		_, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{err: errors.New("denied")}, rec).Execute(context.Background(), job, io.Discard)
+		assert.ErrorContains(t, err, "denied")
+		assert.Empty(t, rec.cmds)
+		assert.NoFileExists(t, overridePath)
+	})
+	t.Run("pull fails", func(t *testing.T) {
+		t.Cleanup(cleanOverride)
+		rec := &recorder{failOn: "pull", failOutput: "pull access denied\n"}
+		_, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard)
+		assert.ErrorContains(t, err, "pull access denied")
+		require.Len(t, rec.cmds, 1, "up must not run")
+	})
+	t.Run("override refused", func(t *testing.T) {
+		other := filepath.Join(stack, "docker-compose.override.yml")
+		require.NoError(t, os.WriteFile(other, []byte("services: {}\n"), 0o644))
+		t.Cleanup(func() { os.Remove(other) })
+		rec := &recorder{}
+		_, err := setup(fakeCompose{rw, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard)
+		assert.ErrorContains(t, err, "refused")
+		assert.Empty(t, rec.cmds)
+	})
+	t.Run("read-only mount", func(t *testing.T) {
+		rec := &recorder{}
+		_, err := setup(fakeCompose{ro, []container.Summary{ct}}, &fakeDist{digest: d}, rec).Execute(context.Background(), job, io.Discard)
+		assert.ErrorContains(t, err, "no stack directory is mounted")
+		assert.Empty(t, rec.cmds)
+		assert.NoFileExists(t, overridePath)
+	})
+}

internal/agent/updater/dispatch.go

@@ -0,0 +1,28 @@
+package updater
+
+import (
+	"context"
+	"fmt"
+	"io"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+type dispatch struct{ compose, swarm Executor }
+
+// NewDispatch returns an Executor that picks compose or swarm by job.Kind.
+// KindContainer and unknown kinds fail.
+func NewDispatch(compose, swarm Executor) Executor {
+	return dispatch{compose: compose, swarm: swarm}
+}
+
+func (d dispatch) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
+	switch job.Kind {
+	case protocol.KindCompose:
+		return d.compose.Execute(ctx, job, out)
+	case protocol.KindSwarm:
+		return d.swarm.Execute(ctx, job, out)
+	default:
+		return "", fmt.Errorf("cannot update a service of kind %q", job.Kind)
+	}
+}

internal/agent/updater/dispatch_test.go

@@ -0,0 +1,31 @@
+package updater
+
+import (
+	"context"
+	"io"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+func TestDispatch(t *testing.T) {
+	compose := execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "compose", nil })
+	swarm := execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "swarm", nil })
+	d := NewDispatch(compose, swarm)
+
+	got, err := d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindCompose}, io.Discard)
+	require.NoError(t, err)
+	assert.Equal(t, "compose", got)
+
+	got, err = d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindSwarm}, io.Discard)
+	require.NoError(t, err)
+	assert.Equal(t, "swarm", got)
+
+	_, err = d.Execute(context.Background(), protocol.JobStart{Kind: protocol.KindContainer}, io.Discard)
+	assert.ErrorContains(t, err, "cannot update")
+	_, err = d.Execute(context.Background(), protocol.JobStart{Kind: "weird"}, io.Discard)
+	assert.ErrorContains(t, err, "cannot update")
+}

internal/agent/updater/override.go

@@ -0,0 +1,192 @@
+package updater
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/mount"
+	"gopkg.in/yaml.v3"
+)
+
+// OverrideName is the override file the agent owns. docker compose loads it
+// automatically next to compose.yaml or docker-compose.yml.
+const OverrideName = "compose.override.yaml"
+
+// overrideHeader marks a file the agent may rewrite. A file without it
+// belongs to the operator and is never touched.
+const overrideHeader = "# managed by package-updater — do not edit\n"
+
+// overrideNames are the override files docker compose auto-loads. It loads
+// only the first one present (in this order), so any other one next to ours
+// would either hide ours or be hidden by it.
+var overrideNames = []string{
+	"compose.override.yml", "compose.override.yaml",
+	"docker-compose.override.yml", "docker-compose.override.yaml",
+}
+
+// WriteOverride sets services.<service>.image to ref in dir/OverrideName,
+// keeping every other entry in the file, and returns the file's path. It
+// refuses when the file exists without the header, or when another
+// auto-loaded override file exists in dir. The write is atomic.
+func WriteOverride(dir, service, ref string) (string, error) {
+	for _, name := range overrideNames {
+		if name == OverrideName {
+			continue
+		}
+		if _, err := os.Lstat(filepath.Join(dir, name)); err == nil {
+			return "", fmt.Errorf("refused: %s exists; the agent only manages %s and never edits operator files", name, OverrideName)
+		} else if !errors.Is(err, os.ErrNotExist) {
+			return "", fmt.Errorf("check %s: %w", name, err)
+		}
+	}
+	path := filepath.Join(dir, OverrideName)
+	doc := map[string]any{}
+	raw, err := os.ReadFile(path)
+	switch {
+	case errors.Is(err, os.ErrNotExist):
+	case err != nil:
+		return "", fmt.Errorf("read override: %w", err)
+	case !bytes.HasPrefix(raw, []byte(overrideHeader)):
+		return "", fmt.Errorf("refused: %s exists and was not written by package-updater", path)
+	default:
+		if err := yaml.Unmarshal(raw, &doc); err != nil {
+			return "", fmt.Errorf("parse override: %w", err)
+		}
+		if doc == nil {
+			doc = map[string]any{}
+		}
+	}
+	services, _ := doc["services"].(map[string]any)
+	if services == nil {
+		services = map[string]any{}
+	}
+	entry, _ := services[service].(map[string]any)
+	if entry == nil {
+		entry = map[string]any{}
+	}
+	entry["image"] = ref
+	services[service] = entry
+	doc["services"] = services
+	body, err := yaml.Marshal(doc)
+	if err != nil {
+		return "", fmt.Errorf("encode override: %w", err)
+	}
+	// Temp file in the same directory + rename: a crash never leaves a
+	// half-written override that compose would then load.
+	tmp, err := os.CreateTemp(dir, "."+OverrideName+".*")
+	if err != nil {
+		return "", fmt.Errorf("write override: %w", err)
+	}
+	defer os.Remove(tmp.Name()) // no-op after a successful rename
+	if _, err := tmp.WriteString(overrideHeader + string(body)); err != nil {
+		tmp.Close()
+		return "", fmt.Errorf("write override: %w", err)
+	}
+	if err := tmp.Chmod(0o644); err != nil {
+		tmp.Close()
+		return "", fmt.Errorf("write override: %w", err)
+	}
+	if err := tmp.Close(); err != nil {
+		return "", fmt.Errorf("write override: %w", err)
+	}
+	if err := os.Rename(tmp.Name(), path); err != nil {
+		return "", fmt.Errorf("write override: %w", err)
+	}
+	return path, nil
+}
+
+// SelfInspector is the subset of the Docker SDK MountRoots needs.
+type SelfInspector interface {
+	ContainerInspect(ctx context.Context, containerID string) (container.InspectResponse, error)
+}
+
+// errNoRoots is returned when the agent has no directory it may write stacks in.
+var errNoRoots = errors.New("refused: no stack directory is mounted into the agent at the same path (read-write); " +
+	"mount each stacks directory as /path:/path and check that SELF_ID (default: the hostname) names the agent's container")
+
+// MountRoots returns the directories the agent may update Compose stacks in:
+// its own container's read-write bind mounts whose host path equals the
+// path inside the container. Compose labels hold host paths, so only those
+// mounts make a label path usable as is.
+func MountRoots(ctx context.Context, cli SelfInspector, selfID string) ([]string, error) {
+	if selfID == "" {
+		return nil, errNoRoots
+	}
+	insp, err := cli.ContainerInspect(ctx, selfID)
+	if err != nil {
+		return nil, fmt.Errorf("inspect own container %s: %w", selfID, err)
+	}
+	var roots []string
+	for _, m := range insp.Mounts {
+		if m.Type == mount.TypeBind && m.RW && filepath.Clean(m.Source) == filepath.Clean(m.Destination) {
+			roots = append(roots, filepath.Clean(m.Destination))
+		}
+	}
+	return roots, nil
+}
+
+// composeFiles returns the -f arguments for a job: the stack's config files
+// from the container label with any copy of the override removed, then the
+// override last so it wins. Every file, and the real location of the stack
+// directory after resolving symlinks, must be inside one of roots.
+func composeFiles(roots []string, configFilesLabel string) (files []string, dir string, err error) {
+	if len(roots) == 0 {
+		return nil, "", errNoRoots
+	}
+	for _, f := range strings.Split(configFilesLabel, ",") {
+		if f = strings.TrimSpace(f); f != "" {
+			files = append(files, f)
+		}
+	}
+	if len(files) == 0 {
+		return nil, "", errors.New("container has no com.docker.compose.project.config_files label")
+	}
+	dir = filepath.Dir(files[0])
+	override := filepath.Join(dir, OverrideName)
+	out := files[:0]
+	for _, f := range files {
+		if !filepath.IsAbs(f) || !insideAny(roots, f) {
+			return nil, "", notMounted(filepath.Dir(f))
+		}
+		if filepath.Clean(f) != override {
+			out = append(out, f)
+		}
+	}
+	real, err := filepath.EvalSymlinks(dir)
+	if err != nil {
+		return nil, "", fmt.Errorf("resolve %s: %w", dir, err)
+	}
+	// Only directories count once resolved: this drops a same-path mount of
+	// a single file such as /var/run/docker.sock.
+	var realRoots []string
+	for _, r := range roots {
+		if rr, err := filepath.EvalSymlinks(r); err == nil {
+			if fi, err := os.Stat(rr); err == nil && fi.IsDir() {
+				realRoots = append(realRoots, rr)
+			}
+		}
+	}
+	if !insideAny(realRoots, real) {
+		return nil, "", notMounted(dir)
+	}
+	return append(out, override), dir, nil
+}
+
+func insideAny(roots []string, path string) bool {
+	for _, r := range roots {
+		if IsInsideRoot(r, path) {
+			return true
+		}
+	}
+	return false
+}
+
+func notMounted(dir string) error {
+	return fmt.Errorf("stack directory %s is not mounted into the agent at the same path (read-write)", dir)
+}

internal/agent/updater/override_test.go

@@ -0,0 +1,199 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/docker/docker/api/types/container"
+	"github.com/docker/docker/api/types/mount"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+	"gopkg.in/yaml.v3"
+)
+
+func readOverride(t *testing.T, dir string) (string, map[string]any) {
+	t.Helper()
+	raw, err := os.ReadFile(filepath.Join(dir, OverrideName))
+	require.NoError(t, err)
+	var doc map[string]any
+	require.NoError(t, yaml.Unmarshal(raw, &doc))
+	return string(raw), doc
+}
+
+func serviceImage(doc map[string]any, name string) any {
+	svcs, _ := doc["services"].(map[string]any)
+	entry, _ := svcs[name].(map[string]any)
+	return entry["image"]
+}
+
+func TestWriteOverride(t *testing.T) {
+	t.Run("new file", func(t *testing.T) {
+		dir := t.TempDir()
+		path, err := WriteOverride(dir, "web", "a:1@sha256:aa")
+		require.NoError(t, err)
+		assert.Equal(t, filepath.Join(dir, OverrideName), path)
+		raw, doc := readOverride(t, dir)
+		assert.True(t, len(raw) > len(overrideHeader) && raw[:len(overrideHeader)] == overrideHeader)
+		assert.Equal(t, "a:1@sha256:aa", serviceImage(doc, "web"))
+	})
+	t.Run("keeps other services", func(t *testing.T) {
+		dir := t.TempDir()
+		require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(overrideHeader+"services:\n  db:\n    image: db:1\n"), 0o644))
+		_, err := WriteOverride(dir, "web", "a:1@sha256:aa")
+		require.NoError(t, err)
+		_, doc := readOverride(t, dir)
+		assert.Equal(t, "db:1", serviceImage(doc, "db"))
+		assert.Equal(t, "a:1@sha256:aa", serviceImage(doc, "web"))
+	})
+	t.Run("replaces own entry", func(t *testing.T) {
+		dir := t.TempDir()
+		require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(overrideHeader+"services:\n  web:\n    image: old\n"), 0o644))
+		_, err := WriteOverride(dir, "web", "a:2@sha256:bb")
+		require.NoError(t, err)
+		_, doc := readOverride(t, dir)
+		assert.Equal(t, "a:2@sha256:bb", serviceImage(doc, "web"))
+	})
+	t.Run("operator file without header is never touched", func(t *testing.T) {
+		dir := t.TempDir()
+		const body = "services:\n  web:\n    environment: [A=1]\n"
+		require.NoError(t, os.WriteFile(filepath.Join(dir, OverrideName), []byte(body), 0o644))
+		_, err := WriteOverride(dir, "web", "a:1@sha256:aa")
+		assert.ErrorContains(t, err, "refused")
+		got, _ := os.ReadFile(filepath.Join(dir, OverrideName))
+		assert.Equal(t, body, string(got))
+	})
+	for _, other := range []string{"compose.override.yml", "docker-compose.override.yml", "docker-compose.override.yaml"} {
+		t.Run("refuses next to "+other, func(t *testing.T) {
+			dir := t.TempDir()
+			require.NoError(t, os.WriteFile(filepath.Join(dir, other), []byte("services: {}\n"), 0o644))
+			_, err := WriteOverride(dir, "web", "a:1@sha256:aa")
+			assert.ErrorContains(t, err, other)
+			_, statErr := os.Stat(filepath.Join(dir, OverrideName))
+			assert.True(t, errors.Is(statErr, os.ErrNotExist))
+		})
+	}
+	t.Run("leaves no temp files", func(t *testing.T) {
+		dir := t.TempDir()
+		_, err := WriteOverride(dir, "web", "a:1@sha256:aa")
+		require.NoError(t, err)
+		entries, err := os.ReadDir(dir)
+		require.NoError(t, err)
+		require.Len(t, entries, 1)
+		assert.Equal(t, OverrideName, entries[0].Name())
+	})
+}
+
+type fakeSelf struct {
+	mounts []container.MountPoint
+	err    error
+}
+
+func (f fakeSelf) ContainerInspect(context.Context, string) (container.InspectResponse, error) {
+	return container.InspectResponse{Mounts: f.mounts}, f.err
+}
+
+func TestMountRoots(t *testing.T) {
+	bind := func(src, dst string, rw bool) container.MountPoint {
+		return container.MountPoint{Type: mount.TypeBind, Source: src, Destination: dst, RW: rw}
+	}
+	cases := []struct {
+		name    string
+		self    fakeSelf
+		selfID  string
+		want    []string
+		wantErr string
+	}{
+		{"rw same path", fakeSelf{mounts: []container.MountPoint{bind("/srv/stacks", "/srv/stacks", true)}}, "me", []string{"/srv/stacks"}, ""},
+		{"read-only", fakeSelf{mounts: []container.MountPoint{bind("/srv/stacks", "/srv/stacks", false)}}, "me", nil, ""},
+		{"path translated", fakeSelf{mounts: []container.MountPoint{bind("/home/x/stacks", "/stacks", true)}}, "me", nil, ""},
+		{"named volume", fakeSelf{mounts: []container.MountPoint{{Type: mount.TypeVolume, Source: "/v", Destination: "/v", RW: true}}}, "me", nil, ""},
+		{"inspect error", fakeSelf{err: errors.New("no such container")}, "me", nil, "inspect own container"},
+		{"no self id", fakeSelf{}, "", nil, "no stack directory is mounted"},
+	}
+	for _, c := range cases {
+		t.Run(c.name, func(t *testing.T) {
+			got, err := MountRoots(context.Background(), c.self, c.selfID)
+			if c.wantErr != "" {
+				assert.ErrorContains(t, err, c.wantErr)
+				return
+			}
+			require.NoError(t, err)
+			assert.Equal(t, c.want, got)
+		})
+	}
+}
+
+func TestComposeFiles(t *testing.T) {
+	root := t.TempDir()
+	root, err := filepath.EvalSymlinks(root) // macOS temp dirs are symlinked
+	require.NoError(t, err)
+	mk := func(p ...string) string {
+		dir := filepath.Join(append([]string{root}, p...)...)
+		require.NoError(t, os.MkdirAll(dir, 0o755))
+		return dir
+	}
+	a := mk("a")
+	nested := mk("apps", "x", "y")
+	second := t.TempDir()
+	second, _ = filepath.EvalSymlinks(second)
+	secondDir := filepath.Join(second, "s")
+	require.NoError(t, os.MkdirAll(secondDir, 0o755))
+
+	outside, _ := filepath.EvalSymlinks(t.TempDir())
+	require.NoError(t, os.Symlink(outside, filepath.Join(root, "evil")))
+
+	sock := filepath.Join(root, "docker.sock")
+	require.NoError(t, os.WriteFile(sock, nil, 0o644))
+
+	t.Run("adds override last", func(t *testing.T) {
+		files, dir, err := composeFiles([]string{root}, filepath.Join(a, "compose.yaml"))
+		require.NoError(t, err)
+		assert.Equal(t, []string{filepath.Join(a, "compose.yaml"), filepath.Join(a, OverrideName)}, files)
+		assert.Equal(t, a, dir)
+	})
+	t.Run("override in label listed once, last", func(t *testing.T) {
+		label := filepath.Join(a, OverrideName) + "," + filepath.Join(a, "compose.yaml") + "," + filepath.Join(a, "extra.yaml")
+		files, _, err := composeFiles([]string{root}, label)
+		require.NoError(t, err)
+		assert.Equal(t, []string{filepath.Join(a, "compose.yaml"), filepath.Join(a, "extra.yaml"), filepath.Join(a, OverrideName)}, files)
+	})
+	t.Run("nested", func(t *testing.T) {
+		_, dir, err := composeFiles([]string{root}, filepath.Join(nested, "compose.yaml"))
+		require.NoError(t, err)
+		assert.Equal(t, nested, dir)
+	})
+	t.Run("second root", func(t *testing.T) {
+		_, _, err := composeFiles([]string{root, second}, filepath.Join(secondDir, "compose.yaml"))
+		require.NoError(t, err)
+	})
+	t.Run("empty label", func(t *testing.T) {
+		_, _, err := composeFiles([]string{root}, "")
+		assert.ErrorContains(t, err, "config_files")
+	})
+	t.Run("no roots", func(t *testing.T) {
+		_, _, err := composeFiles(nil, filepath.Join(a, "compose.yaml"))
+		assert.ErrorContains(t, err, "no stack directory is mounted")
+	})
+	t.Run("outside every root", func(t *testing.T) {
+		_, _, err := composeFiles([]string{root}, filepath.Join(second, "s", "compose.yaml"))
+		assert.EqualError(t, err, "stack directory "+filepath.Join(second, "s")+" is not mounted into the agent at the same path (read-write)")
+	})
+	t.Run("symlink escaping the root", func(t *testing.T) {
+		_, _, err := composeFiles([]string{root}, filepath.Join(root, "evil", "compose.yaml"))
+		assert.EqualError(t, err, "stack directory "+filepath.Join(root, "evil")+" is not mounted into the agent at the same path (read-write)")
+	})
+	t.Run("prefix trick", func(t *testing.T) {
+		evil := root + "-evil"
+		require.NoError(t, os.MkdirAll(evil, 0o755))
+		t.Cleanup(func() { os.RemoveAll(evil) })
+		_, _, err := composeFiles([]string{root}, filepath.Join(evil, "compose.yaml"))
+		assert.ErrorContains(t, err, "not mounted")
+	})
+	t.Run("a file root allows nothing", func(t *testing.T) {
+		_, _, err := composeFiles([]string{sock}, filepath.Join(sock, "compose.yaml"))
+		assert.Error(t, err)
+	})
+}

internal/agent/updater/pathcheck.go

@@ -1,4 +1,4 @@
-package discovery
+package updater
 
 import (
 	"path/filepath"

internal/agent/updater/pathcheck_test.go

@@ -1,9 +1,9 @@
-package discovery_test
+package updater_test
 
 import (
 	"testing"
 
-	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/shcizo/package-updater/internal/agent/updater"
 	"github.com/stretchr/testify/require"
 )
 
@@ -26,7 +26,7 @@ func TestIsInsideRoot(t *testing.T) {
 	}
 	for _, c := range cases {
 		t.Run(c.name, func(t *testing.T) {
-			require.Equal(t, c.want, discovery.IsInsideRoot(c.root, c.path))
+			require.Equal(t, c.want, updater.IsInsideRoot(c.root, c.path))
 		})
 	}
 }

internal/agent/updater/registryauth.go

@@ -0,0 +1,106 @@
+package updater
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"strings"
+
+	"github.com/distribution/reference"
+	"github.com/docker/docker/api/types/registry"
+)
+
+// RegistryAuth holds inline credentials from a docker config.json, already
+// encoded as X-Registry-Auth values. A nil *RegistryAuth means anonymous
+// access.
+type RegistryAuth struct {
+	byHost map[string]string
+}
+
+type dockerConfigFile struct {
+	Auths map[string]struct {
+		Auth          string `json:"auth"`
+		IdentityToken string `json:"identitytoken"`
+	} `json:"auths"`
+	CredsStore  string            `json:"credsStore"`
+	CredHelpers map[string]string `json:"credHelpers"`
+}
+
+// LoadRegistryAuth reads dockerConfigDir/config.json. An empty dir returns
+// nil, nil. Only inline `auth` and `identitytoken` entries are supported;
+// credential helpers are not available inside the agent container, which is
+// logged once.
+func LoadRegistryAuth(dockerConfigDir string, log *slog.Logger) (*RegistryAuth, error) {
+	if dockerConfigDir == "" {
+		return nil, nil
+	}
+	path := filepath.Join(dockerConfigDir, "config.json")
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return nil, fmt.Errorf("read %s: %w", path, err)
+	}
+	var cfg dockerConfigFile
+	if err := json.Unmarshal(raw, &cfg); err != nil {
+		return nil, fmt.Errorf("parse %s: %w", path, err)
+	}
+	if (cfg.CredsStore != "" || len(cfg.CredHelpers) > 0) && log != nil {
+		log.Warn("registry_cred_helpers_unsupported", "path", path)
+	}
+	a := &RegistryAuth{byHost: make(map[string]string)}
+	for key, entry := range cfg.Auths {
+		host := normaliseRegistryKey(key)
+		ac := registry.AuthConfig{ServerAddress: host}
+		switch {
+		case entry.IdentityToken != "":
+			ac.IdentityToken = entry.IdentityToken
+		case entry.Auth != "":
+			dec, err := base64.StdEncoding.DecodeString(entry.Auth)
+			if err != nil {
+				return nil, fmt.Errorf("%s: auth for %s is not base64", path, key)
+			}
+			user, pass, ok := strings.Cut(string(dec), ":")
+			if !ok {
+				return nil, errors.New(path + ": auth for " + key + " is not user:password")
+			}
+			ac.Username, ac.Password = user, pass
+		default:
+			continue
+		}
+		enc, err := registry.EncodeAuthConfig(ac)
+		if err != nil {
+			return nil, fmt.Errorf("encode auth for %s: %w", key, err)
+		}
+		a.byHost[host] = enc
+	}
+	return a, nil
+}
+
+// normaliseRegistryKey reduces a config.json key to a bare registry host,
+// mapping every Docker Hub spelling to "docker.io".
+func normaliseRegistryKey(key string) string {
+	k := strings.TrimPrefix(strings.TrimPrefix(key, "https://"), "http://")
+	if i := strings.Index(k, "/"); i >= 0 {
+		k = k[:i]
+	}
+	switch k {
+	case "index.docker.io", "docker.io", "registry-1.docker.io":
+		return "docker.io"
+	}
+	return k
+}
+
+// For returns the X-Registry-Auth value for repo's registry, or "".
+func (a *RegistryAuth) For(repo string) string {
+	if a == nil {
+		return ""
+	}
+	named, err := reference.ParseNormalizedNamed(repo)
+	if err != nil {
+		return ""
+	}
+	return a.byHost[reference.Domain(named)]
+}

internal/agent/updater/registryauth_test.go

@@ -0,0 +1,72 @@
+package updater
+
+import (
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"log/slog"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/docker/docker/api/types/registry"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func writeDockerConfig(t *testing.T, body string) string {
+	t.Helper()
+	dir := t.TempDir()
+	require.NoError(t, os.WriteFile(filepath.Join(dir, "config.json"), []byte(body), 0o600))
+	return dir
+}
+
+func decodeAuth(t *testing.T, enc string) registry.AuthConfig {
+	t.Helper()
+	raw, err := base64.URLEncoding.DecodeString(enc)
+	require.NoError(t, err)
+	var ac registry.AuthConfig
+	require.NoError(t, json.Unmarshal(raw, &ac))
+	return ac
+}
+
+func TestRegistryAuth(t *testing.T) {
+	user := base64.StdEncoding.EncodeToString([]byte("bob:s3cret"))
+	log := slog.New(slog.NewTextHandler(io.Discard, nil))
+
+	t.Run("docker hub spellings", func(t *testing.T) {
+		for _, key := range []string{"https://index.docker.io/v1/", "index.docker.io", "docker.io", "registry-1.docker.io"} {
+			dir := writeDockerConfig(t, `{"auths":{"`+key+`":{"auth":"`+user+`"}}}`)
+			a, err := LoadRegistryAuth(dir, log)
+			require.NoError(t, err)
+			for _, repo := range []string{"gitea/gitea", "docker.io/library/nginx", "nginx"} {
+				ac := decodeAuth(t, a.For(repo))
+				assert.Equal(t, "bob", ac.Username, "%s for %s", key, repo)
+				assert.Equal(t, "s3cret", ac.Password)
+			}
+			assert.Empty(t, a.For("gitea.shcizo.se/shcizo/slusk"))
+		}
+	})
+	t.Run("custom registry with scheme", func(t *testing.T) {
+		dir := writeDockerConfig(t, `{"auths":{"https://gitea.shcizo.se":{"auth":"`+user+`"},"ghcr.io":{"identitytoken":"tok"}}}`)
+		a, err := LoadRegistryAuth(dir, log)
+		require.NoError(t, err)
+		assert.Equal(t, "bob", decodeAuth(t, a.For("gitea.shcizo.se/shcizo/slusk")).Username)
+		assert.Equal(t, "tok", decodeAuth(t, a.For("ghcr.io/x/y")).IdentityToken)
+		assert.Empty(t, a.For("registry.example.com/x"))
+	})
+	t.Run("empty dir is anonymous", func(t *testing.T) {
+		a, err := LoadRegistryAuth("", log)
+		require.NoError(t, err)
+		assert.Nil(t, a)
+		assert.Empty(t, a.For("gitea/gitea"))
+	})
+	t.Run("malformed json", func(t *testing.T) {
+		_, err := LoadRegistryAuth(writeDockerConfig(t, "{"), log)
+		assert.Error(t, err)
+	})
+	t.Run("missing file", func(t *testing.T) {
+		_, err := LoadRegistryAuth(t.TempDir(), log)
+		assert.Error(t, err)
+	})
+}

internal/agent/updater/resolve.go

@@ -0,0 +1,49 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/docker/docker/api/types/registry"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+)
+
+// DistributionClient is the subset of the Docker SDK the Resolver needs.
+// *client.Client satisfies it.
+type DistributionClient interface {
+	DistributionInspect(ctx context.Context, imageRef, encodedRegistryAuth string) (registry.DistributionInspect, error)
+}
+
+// Resolver turns "repo:tag" into "repo:tag@digest" by asking the local
+// daemon, which contacts the registry.
+type Resolver struct {
+	cli  DistributionClient
+	auth *RegistryAuth
+}
+
+// NewResolver returns a Resolver. auth may be nil (anonymous access).
+func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver {
+	return &Resolver{cli: cli, auth: auth}
+}
+
+// Resolve returns ref pinned to the digest its tag currently points at. It
+// fails for a ref without a tag or with a digest already.
+func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error) {
+	p := imageref.Parse(ref)
+	switch {
+	case p.Digest != "":
+		return "", fmt.Errorf("image %q already has a digest", ref)
+	case p.Tag == "":
+		return "", fmt.Errorf("image %q has no tag to resolve", ref)
+	}
+	info, err := r.cli.DistributionInspect(ctx, ref, r.auth.For(p.Repo))
+	if err != nil {
+		return "", fmt.Errorf("resolve %s: %w", ref, err)
+	}
+	if info.Descriptor.Digest == "" {
+		return "", errors.New("resolve " + ref + ": registry returned no digest")
+	}
+	return ref + "@" + info.Descriptor.Digest.String(), nil
+}

internal/agent/updater/resolve_test.go

@@ -0,0 +1,38 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func TestResolver(t *testing.T) {
+	const d = "sha256:d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1d1"
+	auth := &RegistryAuth{byHost: map[string]string{"gitea.shcizo.se": "enc-gitea"}}
+
+	t.Run("resolves tag with registry auth", func(t *testing.T) {
+		dist := &fakeDist{digest: d}
+		got, err := NewResolver(dist, auth).Resolve(context.Background(), "gitea.shcizo.se/shcizo/slusk:1.4")
+		require.NoError(t, err)
+		assert.Equal(t, "gitea.shcizo.se/shcizo/slusk:1.4@"+d, got)
+		assert.Equal(t, []string{"gitea.shcizo.se/shcizo/slusk:1.4"}, dist.refs)
+		assert.Equal(t, []string{"enc-gitea"}, dist.auths)
+	})
+	t.Run("digest or missing tag is refused without asking the daemon", func(t *testing.T) {
+		for _, ref := range []string{"repo@sha256:x", "repo:1@sha256:x", "repo"} {
+			dist := &fakeDist{digest: d}
+			_, err := NewResolver(dist, nil).Resolve(context.Background(), ref)
+			assert.Error(t, err, ref)
+			assert.Empty(t, dist.refs)
+		}
+	})
+	t.Run("daemon error is wrapped", func(t *testing.T) {
+		cause := errors.New("unauthorized")
+		_, err := NewResolver(&fakeDist{err: cause}, nil).Resolve(context.Background(), "repo:1")
+		assert.ErrorIs(t, err, cause)
+		assert.ErrorContains(t, err, "repo:1")
+	})
+}

internal/agent/updater/runner.go

@@ -0,0 +1,272 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"sort"
+	"sync"
+	"time"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// maxProgress caps the output buffered between two TakeProgress calls, so a
+// chatty job cannot grow a single job.progress message without bound.
+const maxProgress = 32 << 10
+
+// Executor deploys one job and returns the "repo:tag@digest" it deployed.
+// Everything it writes to out is streamed to the master as job output.
+type Executor interface {
+	Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error)
+}
+
+// Progress is output a running job produced since the previous TakeProgress.
+type Progress struct{ ID, Output string }
+
+// Finished is a job's outcome, kept until the master acknowledges it.
+type Finished struct {
+	ID     string
+	Result protocol.JobResult
+}
+
+// tailBuffer keeps the last max bytes written to it.
+type tailBuffer struct {
+	max       int
+	buf       []byte
+	truncated bool
+}
+
+func (b *tailBuffer) Write(p []byte) {
+	b.buf = append(b.buf, p...)
+	if over := len(b.buf) - b.max; over > 0 {
+		b.buf = append(b.buf[:0], b.buf[over:]...)
+		b.truncated = true
+	}
+}
+
+func (b *tailBuffer) String() string  { return string(b.buf) }
+func (b *tailBuffer) Truncated() bool { return b.truncated }
+func (b *tailBuffer) Reset()          { b.buf, b.truncated = nil, false }
+
+type jobState struct {
+	start    protocol.JobStart
+	running  bool
+	result   *protocol.JobResult
+	output   tailBuffer // everything, capped at protocol.MaxJobOutput
+	unsent   tailBuffer // since the last TakeProgress, capped at maxProgress
+	reported bool       // a Progress has been taken since the job started
+	finished time.Time  // orders Pending
+}
+
+// Runner executes jobs one at a time in the order they arrive and keeps each
+// job's result until Ack. Two docker compose calls against the same stack
+// would race, so there is deliberately exactly one worker.
+//
+// Runner satisfies conn.Jobs.
+type Runner struct {
+	exec    Executor
+	timeout time.Duration
+	now     func() time.Time
+
+	mu    sync.Mutex // guards jobs and queue
+	jobs  map[string]*jobState
+	queue []string
+
+	wake    chan struct{} // wakes the worker
+	changed chan struct{} // tells the connection there is something to send
+}
+
+// NewRunner returns a Runner that gives every job at most timeout. Call Run
+// to start the worker.
+func NewRunner(exec Executor, timeout time.Duration) *Runner {
+	return &Runner{
+		exec:    exec,
+		timeout: timeout,
+		now:     time.Now,
+		jobs:    make(map[string]*jobState),
+		wake:    make(chan struct{}, 1),
+		changed: make(chan struct{}, 1),
+	}
+}
+
+// signal does a non-blocking send so neither side ever waits on the other.
+func signal(ch chan struct{}) {
+	select {
+	case ch <- struct{}{}:
+	default:
+	}
+}
+
+// Submit queues a job. A job whose ID the runner already holds (queued,
+// running, or finished and not yet acknowledged) is ignored.
+func (r *Runner) Submit(id string, start protocol.JobStart) {
+	r.mu.Lock()
+	if _, ok := r.jobs[id]; ok {
+		r.mu.Unlock()
+		return
+	}
+	r.jobs[id] = &jobState{
+		start:  start,
+		output: tailBuffer{max: protocol.MaxJobOutput},
+		unsent: tailBuffer{max: maxProgress},
+	}
+	r.queue = append(r.queue, id)
+	r.mu.Unlock()
+	signal(r.wake)
+}
+
+// Ack forgets a finished job. Acks for unknown, queued or running jobs are
+// ignored.
+func (r *Runner) Ack(id string) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if j, ok := r.jobs[id]; ok && j.result != nil {
+		delete(r.jobs, id)
+	}
+}
+
+// Known returns the sorted IDs of every job the runner holds.
+func (r *Runner) Known() []string {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	ids := make([]string, 0, len(r.jobs))
+	for id := range r.jobs {
+		ids = append(ids, id)
+	}
+	sort.Strings(ids)
+	return ids
+}
+
+// TakeProgress returns the output running jobs produced since the previous
+// call. A job's first entry (possibly with empty Output) tells the master it
+// is running.
+func (r *Runner) TakeProgress() []Progress {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []Progress
+	for id, j := range r.jobs {
+		if !j.running {
+			continue
+		}
+		s := j.unsent.String()
+		if j.reported && s == "" {
+			continue
+		}
+		j.reported = true
+		j.unsent.Reset()
+		out = append(out, Progress{ID: id, Output: s})
+	}
+	sort.Slice(out, func(a, b int) bool { return out[a].ID < out[b].ID })
+	return out
+}
+
+// Pending returns the results not yet acknowledged, oldest first.
+func (r *Runner) Pending() []Finished {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	var out []Finished
+	var at []time.Time
+	for id, j := range r.jobs {
+		if j.result != nil {
+			out = append(out, Finished{ID: id, Result: *j.result})
+			at = append(at, j.finished)
+		}
+	}
+	idx := make([]int, len(out))
+	for i := range idx {
+		idx[i] = i
+	}
+	sort.Slice(idx, func(a, b int) bool {
+		if !at[idx[a]].Equal(at[idx[b]]) {
+			return at[idx[a]].Before(at[idx[b]])
+		}
+		return out[idx[a]].ID < out[idx[b]].ID
+	})
+	sorted := make([]Finished, len(out))
+	for i, k := range idx {
+		sorted[i] = out[k]
+	}
+	return sorted
+}
+
+// Changed returns a channel that receives when there is new progress or a
+// new result to send.
+func (r *Runner) Changed() <-chan struct{} { return r.changed }
+
+// Run works through the queue until ctx ends. Jobs still queued then are
+// dropped; the master marks them lost on the agent's next hello.
+func (r *Runner) Run(ctx context.Context) {
+	for {
+		id, start, ok := r.next()
+		if !ok {
+			select {
+			case <-ctx.Done():
+				return
+			case <-r.wake:
+				continue
+			}
+		}
+		if ctx.Err() != nil {
+			return
+		}
+		r.runOne(ctx, id, start)
+	}
+}
+
+// next pops the next queued job and marks it running.
+func (r *Runner) next() (string, protocol.JobStart, bool) {
+	r.mu.Lock()
+	defer r.mu.Unlock()
+	if len(r.queue) == 0 {
+		return "", protocol.JobStart{}, false
+	}
+	id := r.queue[0]
+	r.queue = r.queue[1:]
+	j := r.jobs[id]
+	j.running = true
+	signal(r.changed)
+	return id, j.start, true
+}
+
+func (r *Runner) runOne(ctx context.Context, id string, start protocol.JobStart) {
+	started := r.now()
+	jobCtx, cancel := context.WithTimeout(ctx, r.timeout)
+	ref, err := r.exec.Execute(jobCtx, start, &jobWriter{r: r, id: id})
+	timedOut := errors.Is(jobCtx.Err(), context.DeadlineExceeded)
+	cancel()
+
+	res := protocol.JobResult{Status: protocol.ResultSucceeded, Ref: ref, StartedAt: started, FinishedAt: r.now()}
+	switch {
+	case timedOut:
+		res = protocol.JobResult{Status: protocol.ResultFailed, Error: fmt.Sprintf("timed out after %s", r.timeout), StartedAt: started, FinishedAt: res.FinishedAt}
+	case err != nil:
+		res = protocol.JobResult{Status: protocol.ResultFailed, Error: err.Error(), StartedAt: started, FinishedAt: res.FinishedAt}
+	}
+
+	r.mu.Lock()
+	j := r.jobs[id]
+	res.Output, res.OutputTruncated = j.output.String(), j.output.Truncated()
+	j.running, j.result, j.finished = false, &res, res.FinishedAt
+	j.unsent.Reset()
+	r.mu.Unlock()
+	signal(r.changed)
+}
+
+// jobWriter feeds an executor's output into the job's buffers.
+type jobWriter struct {
+	r  *Runner
+	id string
+}
+
+func (w *jobWriter) Write(p []byte) (int, error) {
+	w.r.mu.Lock()
+	if j, ok := w.r.jobs[w.id]; ok {
+		j.output.Write(p)
+		j.unsent.Write(p)
+	}
+	w.r.mu.Unlock()
+	signal(w.r.changed)
+	return len(p), nil
+}

internal/agent/updater/runner_test.go

@@ -0,0 +1,200 @@
+package updater
+
+import (
+	"context"
+	"errors"
+	"io"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
+)
+
+// fakeExec writes "pulling <ref>" and then waits on gate (if set). Refs
+// starting with "bad" fail. It records the highest concurrency it saw.
+type fakeExec struct {
+	gate    chan struct{}
+	cur     atomic.Int32
+	maxSeen atomic.Int32
+	started chan string
+}
+
+func newFakeExec() *fakeExec {
+	return &fakeExec{gate: make(chan struct{}), started: make(chan string, 10)}
+}
+
+func (f *fakeExec) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
+	n := f.cur.Add(1)
+	defer f.cur.Add(-1)
+	for {
+		m := f.maxSeen.Load()
+		if n <= m || f.maxSeen.CompareAndSwap(m, n) {
+			break
+		}
+	}
+	f.started <- job.Ref
+	io.WriteString(out, "pulling "+job.Ref+"\n")
+	select {
+	case <-f.gate:
+	case <-ctx.Done():
+		return "", ctx.Err()
+	}
+	if strings.HasPrefix(job.Ref, "bad") {
+		return "", errors.New("boom")
+	}
+	return job.Ref + "@sha256:aa", nil
+}
+
+func startRunner(t *testing.T, exec Executor, timeout time.Duration) *Runner {
+	t.Helper()
+	r := NewRunner(exec, timeout)
+	ctx, cancel := context.WithCancel(context.Background())
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() { defer wg.Done(); r.Run(ctx) }()
+	t.Cleanup(func() { cancel(); wg.Wait() })
+	return r
+}
+
+func waitPending(t *testing.T, r *Runner, n int) []Finished {
+	t.Helper()
+	var got []Finished
+	require.Eventually(t, func() bool { got = r.Pending(); return len(got) >= n }, 2*time.Second, 5*time.Millisecond)
+	return got
+}
+
+func TestRunner_FIFOSingleWorker(t *testing.T) {
+	fe := newFakeExec()
+	r := startRunner(t, fe, time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	r.Submit("2", protocol.JobStart{Ref: "bad:2"})
+	assert.Equal(t, "a:1", <-fe.started)
+	fe.gate <- struct{}{}
+	assert.Equal(t, "bad:2", <-fe.started)
+	fe.gate <- struct{}{}
+
+	got := waitPending(t, r, 2)
+	assert.Equal(t, int32(1), fe.maxSeen.Load())
+	require.Len(t, got, 2)
+	assert.Equal(t, "1", got[0].ID)
+	assert.Equal(t, protocol.ResultSucceeded, got[0].Result.Status)
+	assert.Equal(t, "a:1@sha256:aa", got[0].Result.Ref)
+	assert.Equal(t, "pulling a:1\n", got[0].Result.Output)
+	assert.Equal(t, "2", got[1].ID)
+	assert.Equal(t, protocol.ResultFailed, got[1].Result.Status)
+	assert.Equal(t, "boom", got[1].Result.Error)
+	assert.False(t, got[0].Result.StartedAt.IsZero())
+}
+
+func TestRunner_DuplicateSubmitIgnored(t *testing.T) {
+	fe := newFakeExec()
+	r := startRunner(t, fe, time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	assert.Equal(t, "a:1", <-fe.started)
+	r.Submit("1", protocol.JobStart{Ref: "other:9"})
+	fe.gate <- struct{}{}
+	got := waitPending(t, r, 1)
+	assert.Equal(t, "a:1@sha256:aa", got[0].Result.Ref)
+	select {
+	case ref := <-fe.started:
+		t.Fatalf("duplicate job ran: %s", ref)
+	case <-time.After(50 * time.Millisecond):
+	}
+}
+
+func TestRunner_KnownAndAck(t *testing.T) {
+	fe := newFakeExec()
+	r := startRunner(t, fe, time.Minute)
+	r.Submit("2", protocol.JobStart{Ref: "a:2"})
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	assert.Equal(t, "a:2", <-fe.started)
+	assert.Equal(t, []string{"1", "2"}, r.Known())
+
+	r.Ack("2") // running: ignored
+	assert.Equal(t, []string{"1", "2"}, r.Known())
+
+	fe.gate <- struct{}{}
+	waitPending(t, r, 1)
+	r.Ack("2")
+	assert.Equal(t, []string{"1"}, r.Known())
+	<-fe.started
+	fe.gate <- struct{}{}
+	waitPending(t, r, 1)
+}
+
+func TestRunner_TakeProgress(t *testing.T) {
+	fe := newFakeExec()
+	r := startRunner(t, fe, time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	<-fe.started
+	require.Eventually(t, func() bool { return len(r.Pending()) == 0 && len(r.Known()) == 1 }, time.Second, 5*time.Millisecond)
+
+	var first []Progress
+	require.Eventually(t, func() bool { first = r.TakeProgress(); return len(first) > 0 }, time.Second, 5*time.Millisecond)
+	assert.Equal(t, []Progress{{ID: "1", Output: "pulling a:1\n"}}, first)
+	assert.Empty(t, r.TakeProgress(), "nothing new since the last call")
+
+	fe.gate <- struct{}{}
+	waitPending(t, r, 1)
+	assert.Empty(t, r.TakeProgress(), "a finished job no longer reports progress")
+}
+
+func TestRunner_TakeProgressReportsStartedJobOnce(t *testing.T) {
+	gate := make(chan struct{})
+	started := make(chan struct{})
+	r := startRunner(t, execFunc(func(ctx context.Context, _ protocol.JobStart, _ io.Writer) (string, error) {
+		close(started)
+		<-gate
+		return "x", nil
+	}), time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	<-started
+	assert.Equal(t, []Progress{{ID: "1", Output: ""}}, r.TakeProgress())
+	assert.Empty(t, r.TakeProgress())
+	close(gate)
+}
+
+type execFunc func(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error)
+
+func (f execFunc) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
+	return f(ctx, job, out)
+}
+
+func TestRunner_Timeout(t *testing.T) {
+	r := startRunner(t, execFunc(func(ctx context.Context, _ protocol.JobStart, _ io.Writer) (string, error) {
+		<-ctx.Done()
+		return "", ctx.Err()
+	}), 20*time.Millisecond)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	got := waitPending(t, r, 1)
+	assert.Equal(t, protocol.ResultFailed, got[0].Result.Status)
+	assert.Equal(t, "timed out after 20ms", got[0].Result.Error)
+}
+
+func TestRunner_OutputCapped(t *testing.T) {
+	r := startRunner(t, execFunc(func(_ context.Context, _ protocol.JobStart, out io.Writer) (string, error) {
+		io.WriteString(out, strings.Repeat("a", 10))
+		io.WriteString(out, strings.Repeat("b", protocol.MaxJobOutput))
+		return "x", nil
+	}), time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	got := waitPending(t, r, 1)
+	assert.Equal(t, strings.Repeat("b", protocol.MaxJobOutput), got[0].Result.Output)
+	assert.True(t, got[0].Result.OutputTruncated)
+}
+
+func TestRunner_ChangedSignals(t *testing.T) {
+	r := startRunner(t, execFunc(func(context.Context, protocol.JobStart, io.Writer) (string, error) { return "x", nil }), time.Minute)
+	r.Submit("1", protocol.JobStart{Ref: "a:1"})
+	select {
+	case <-r.Changed():
+	case <-time.After(time.Second):
+		t.Fatal("no change signal")
+	}
+}

internal/agent/updater/swarm.go

@@ -3,14 +3,19 @@ package updater
 import (
 	"context"
 	"fmt"
+	"io"
 
 	"github.com/docker/docker/api/types/swarm"
-	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/docker/docker/api/types/system"
+
+	"github.com/shcizo/package-updater/internal/imageref"
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
-// SwarmDockerClient is the subset of the Docker SDK SwarmExecutor depends
+// SwarmClient is the subset of the Docker SDK SwarmExecutor depends on.
-// on. Defined as an interface so tests can supply a fake.
+// Defined as an interface so tests can supply a fake.
-type SwarmDockerClient interface {
+type SwarmClient interface {
+	Info(ctx context.Context) (system.Info, error)
 	ServiceInspectWithRaw(ctx context.Context, serviceID string, opts swarm.ServiceInspectOptions) (swarm.Service, []byte, error)
 	ServiceUpdate(ctx context.Context, serviceID string, version swarm.Version, service swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error)
 }
@@ -18,35 +23,71 @@ type SwarmDockerClient interface {
 // SwarmExecutor updates a Swarm service's image via the Docker API,
 // equivalent to `docker service update --image`.
 type SwarmExecutor struct {
-	cli SwarmDockerClient
+	cli     SwarmClient
+	resolve *Resolver
+	auth    *RegistryAuth
 }
 
 // NewSwarmExecutor returns an executor that drives Swarm service updates
-// through the Docker API.
+// through the Docker API. auth may be nil (anonymous registry access).
-func NewSwarmExecutor(cli SwarmDockerClient) *SwarmExecutor {
+func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *SwarmExecutor {
-	return &SwarmExecutor{cli: cli}
+	return &SwarmExecutor{cli: cli, resolve: resolve, auth: auth}
 }
 
-// Execute inspects the service to get its current spec and version (required
+// Execute checks that the local daemon is a manager of the job's cluster,
-// by the Docker API as an optimistic-concurrency token), sets the new image
+// pins job.Ref to its current digest and updates the service to it. The
-// on the container spec, and calls ServiceUpdate. QueryRegistry is set so a
+// update is addressed by the ID and version of the service that was
-// floating tag (e.g. ":latest") resolves to a fresh digest and actually
+// inspected, so it changes exactly that object. Success means the daemon
-// triggers a rolling update instead of being treated as unchanged.
+// accepted the new spec, not that the tasks have converged. It returns the
-func (e *SwarmExecutor) Execute(ctx context.Context, job discovery.Job) error {
+// pinned ref.
-	if job.Refused {
+func (e *SwarmExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error) {
-		return fmt.Errorf("refused: %s", job.RefusedReason)
+	info, err := e.cli.Info(ctx)
+	if err != nil {
+		return "", fmt.Errorf("docker info: %w", err)
+	}
+	if !info.Swarm.ControlAvailable {
+		return "", fmt.Errorf("refused: this agent's daemon is not a swarm manager")
+	}
+	actual := ""
+	if info.Swarm.Cluster != nil {
+		actual = info.Swarm.Cluster.ID
+	}
+	if actual != job.SwarmClusterID {
+		return "", fmt.Errorf("refused: job is for swarm cluster %s, this daemon is in %s", job.SwarmClusterID, actual)
 	}
 
-	svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.ServiceID, swarm.ServiceInspectOptions{})
+	svc, _, err := e.cli.ServiceInspectWithRaw(ctx, job.SwarmService, swarm.ServiceInspectOptions{})
 	if err != nil {
-		return fmt.Errorf("inspect service %s: %w", job.ServiceID, err)
+		return "", fmt.Errorf("inspect service %s: %w", job.SwarmService, err)
 	}
+	if svc.Spec.Name != job.SwarmService {
+		return "", fmt.Errorf("inspect service %s: daemon returned service %q", job.SwarmService, svc.Spec.Name)
+	}
+	if svc.Spec.TaskTemplate.ContainerSpec == nil {
+		return "", fmt.Errorf("service %s has no container spec", job.SwarmService)
+	}
+
+	pinned, err := e.resolve.Resolve(ctx, job.Ref)
+	if err != nil {
+		return "", err
+	}
+	fmt.Fprintf(out, "resolved %s\n", pinned)
 
 	spec := svc.Spec
-	spec.TaskTemplate.ContainerSpec.Image = job.Image
+	spec.TaskTemplate.ContainerSpec.Image = pinned
-
+	opts := swarm.ServiceUpdateOptions{QueryRegistry: true}
-	if _, err := e.cli.ServiceUpdate(ctx, job.ServiceID, svc.Version, spec, swarm.ServiceUpdateOptions{QueryRegistry: true}); err != nil {
+	if enc := e.auth.For(imageref.Parse(job.Ref).Repo); enc != "" {
-		return fmt.Errorf("update service %s: %w", job.ServiceID, err)
+		opts.EncodedRegistryAuth = enc
+	} else {
+		opts.RegistryAuthFrom = swarm.RegistryAuthFromSpec
 	}
-	return nil
+	resp, err := e.cli.ServiceUpdate(ctx, svc.ID, svc.Version, spec, opts)
+	if err != nil {
+		return "", fmt.Errorf("update service %s: %w", job.SwarmService, err)
+	}
+	for _, w := range resp.Warnings {
+		fmt.Fprintf(out, "warning: %s\n", w)
+	}
+	fmt.Fprintf(out, "service %s updated\n", job.SwarmService)
+	return pinned, nil
 }

internal/agent/updater/swarm_test.go

@@ -1,99 +1,142 @@
-package updater_test
+package updater
 
 import (
+	"bytes"
 	"context"
 	"errors"
 	"testing"
 
+	"github.com/docker/docker/api/types/registry"
 	"github.com/docker/docker/api/types/swarm"
-	"github.com/shcizo/package-updater/internal/discovery"
+	"github.com/docker/docker/api/types/system"
-	"github.com/shcizo/package-updater/internal/updater"
+	"github.com/opencontainers/go-digest"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	"github.com/shcizo/package-updater/internal/protocol"
 )
 
-type fakeSwarmDockerClient struct {
+type fakeSwarm struct {
-	inspectService swarm.Service
+	info       system.Info
-	inspectErr     error
+	svc        swarm.Service
-	updateErr      error
+	inspectErr error
+	updateErr  error
+	warnings   []string
 
-	inspectCalled         bool
+	inspectCalls int
-	updateCalled          bool
+	updateCalls  int
-	updateCalledServiceID string
+	updID        string
-	updateCalledVersion   swarm.Version
+	updVersion   swarm.Version
-	updateCalledSpec      swarm.ServiceSpec
+	updSpec      swarm.ServiceSpec
-	updateCalledOpts      swarm.ServiceUpdateOptions
+	updOpts      swarm.ServiceUpdateOptions
 }
 
-func (f *fakeSwarmDockerClient) ServiceInspectWithRaw(_ context.Context, _ string, _ swarm.ServiceInspectOptions) (swarm.Service, []byte, error) {
+func (f *fakeSwarm) Info(context.Context) (system.Info, error) { return f.info, nil }
-	f.inspectCalled = true
+
-	return f.inspectService, nil, f.inspectErr
+func (f *fakeSwarm) ServiceInspectWithRaw(context.Context, string, swarm.ServiceInspectOptions) (swarm.Service, []byte, error) {
+	f.inspectCalls++
+	return f.svc, nil, f.inspectErr
 }
 
-func (f *fakeSwarmDockerClient) ServiceUpdate(_ context.Context, serviceID string, version swarm.Version, spec swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) {
+func (f *fakeSwarm) ServiceUpdate(_ context.Context, id string, v swarm.Version, spec swarm.ServiceSpec, opts swarm.ServiceUpdateOptions) (swarm.ServiceUpdateResponse, error) {
-	f.updateCalled = true
+	f.updateCalls++
-	f.updateCalledServiceID = serviceID
+	f.updID, f.updVersion, f.updSpec, f.updOpts = id, v, spec, opts
-	f.updateCalledVersion = version
+	return swarm.ServiceUpdateResponse{Warnings: f.warnings}, f.updateErr
-	f.updateCalledSpec = spec
-	f.updateCalledOpts = opts
-	return swarm.ServiceUpdateResponse{}, f.updateErr
 }
 
-func mkInspectService(version uint64, image string) swarm.Service {
+type fakeDist struct {
+	digest string
+	err    error
+	refs   []string
+	auths  []string
+}
+
+func (f *fakeDist) DistributionInspect(_ context.Context, ref, auth string) (registry.DistributionInspect, error) {
+	f.refs = append(f.refs, ref)
+	f.auths = append(f.auths, auth)
+	var d registry.DistributionInspect
+	d.Descriptor.Digest = digest.Digest(f.digest)
+	return d, f.err
+}
+
+func managerOf(cluster string) system.Info {
+	var info system.Info
+	info.Swarm.ControlAvailable = true
+	info.Swarm.Cluster = &swarm.ClusterInfo{ID: cluster}
+	return info
+}
+
+func swarmSvc(name string) swarm.Service {
 	return swarm.Service{
-		ID:   "svc-myapp",
+		ID:   "svc-id",
-		Meta: swarm.Meta{Version: swarm.Version{Index: version}},
+		Meta: swarm.Meta{Version: swarm.Version{Index: 7}},
 		Spec: swarm.ServiceSpec{
-			Annotations:  swarm.Annotations{Name: "myapp_web"},
+			Annotations:  swarm.Annotations{Name: name},
-			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: image}},
+			TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "gitea/gitea:1.26"}},
 		},
 	}
 }
 
-func TestSwarmExecutor_UpdatesImageAndVersion(t *testing.T) {
+func TestSwarmExecutor(t *testing.T) {
-	fake := &fakeSwarmDockerClient{inspectService: mkInspectService(7, "registry.example.com/myapp:v1")}
+	job := protocol.JobStart{Kind: protocol.KindSwarm, SwarmClusterID: "c1", SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"}
-	e := updater.NewSwarmExecutor(fake)
+	workerInfo := managerOf("c1")
+	workerInfo.Swarm.ControlAvailable = false
+	noContainer := swarmSvc("gitea_gitea")
+	noContainer.Spec.TaskTemplate.ContainerSpec = nil
 
-	job := discovery.Job{Service: "myapp_web", ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"}
+	cases := []struct {
-	err := e.Execute(context.Background(), job)
+		name        string
-
+		fake        *fakeSwarm
-	require.NoError(t, err)
+		auth        *RegistryAuth
-	require.True(t, fake.updateCalled)
+		wantErr     string
-	require.Equal(t, "svc-myapp", fake.updateCalledServiceID)
+		wantUpdate  bool
-	require.Equal(t, uint64(7), fake.updateCalledVersion.Index)
+		wantEncAuth bool
-	require.Equal(t, "registry.example.com/myapp:v2", fake.updateCalledSpec.TaskTemplate.ContainerSpec.Image)
+	}{
-	require.True(t, fake.updateCalledOpts.QueryRegistry)
+		{name: "happy path, no auth", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea")}, wantUpdate: true},
-}
+		{name: "happy path, auth known", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea"), warnings: []string{"w1"}},
-
+			auth: &RegistryAuth{byHost: map[string]string{"docker.io": "enc"}}, wantUpdate: true, wantEncAuth: true},
-func TestSwarmExecutor_InspectError(t *testing.T) {
+		{name: "cluster mismatch", fake: &fakeSwarm{info: managerOf("other"), svc: swarmSvc("gitea_gitea")},
-	fake := &fakeSwarmDockerClient{inspectErr: errors.New("service not found")}
+			wantErr: "refused: job is for swarm cluster c1, this daemon is in other"},
-	e := updater.NewSwarmExecutor(fake)
+		{name: "worker node", fake: &fakeSwarm{info: workerInfo, svc: swarmSvc("gitea_gitea")}, wantErr: "not a swarm manager"},
-
+		{name: "other service returned", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_other")}, wantErr: "daemon returned service"},
-	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-missing", Image: "x:v2"})
+		{name: "no container spec", fake: &fakeSwarm{info: managerOf("c1"), svc: noContainer}, wantErr: "no container spec"},
-
+		{name: "inspect fails", fake: &fakeSwarm{info: managerOf("c1"), inspectErr: errors.New("gone")}, wantErr: "gone"},
-	require.Error(t, err)
+		{name: "update fails", fake: &fakeSwarm{info: managerOf("c1"), svc: swarmSvc("gitea_gitea"), updateErr: errors.New("rejected")},
-	require.False(t, fake.updateCalled)
+			wantErr: "rejected", wantUpdate: true},
-}
+	}
-
+	for _, c := range cases {
-func TestSwarmExecutor_UpdateError(t *testing.T) {
+		t.Run(c.name, func(t *testing.T) {
-	fake := &fakeSwarmDockerClient{
+			dist := &fakeDist{digest: "sha256:" + string(bytes.Repeat([]byte("a"), 64))}
-		inspectService: mkInspectService(1, "registry.example.com/myapp:v1"),
+			e := NewSwarmExecutor(c.fake, NewResolver(dist, c.auth), c.auth)
-		updateErr:      errors.New("update rejected"),
+			var out bytes.Buffer
+			ref, err := e.Execute(context.Background(), job, &out)
+			if c.wantErr != "" {
+				assert.ErrorContains(t, err, c.wantErr)
+			} else {
+				require.NoError(t, err)
+				assert.Equal(t, "gitea/gitea:1.27@"+dist.digest, ref)
+				assert.Contains(t, out.String(), "resolved "+ref)
+			}
+			if c.wantErr == "refused: job is for swarm cluster c1, this daemon is in other" || c.wantErr == "not a swarm manager" {
+				assert.Zero(t, c.fake.inspectCalls)
+			}
+			if !c.wantUpdate {
+				assert.Zero(t, c.fake.updateCalls)
+				return
+			}
+			assert.Equal(t, 1, c.fake.updateCalls)
+			assert.Equal(t, "svc-id", c.fake.updID)
+			assert.Equal(t, uint64(7), c.fake.updVersion.Index)
+			assert.Equal(t, "gitea/gitea:1.27@"+dist.digest, c.fake.updSpec.TaskTemplate.ContainerSpec.Image)
+			assert.True(t, c.fake.updOpts.QueryRegistry)
+			if c.wantEncAuth {
+				assert.Equal(t, "enc", c.fake.updOpts.EncodedRegistryAuth)
+				assert.Empty(t, c.fake.updOpts.RegistryAuthFrom)
+				assert.Contains(t, out.String(), "warning: w1")
+			} else {
+				assert.Empty(t, c.fake.updOpts.EncodedRegistryAuth)
+				assert.Equal(t, swarm.RegistryAuthFromSpec, c.fake.updOpts.RegistryAuthFrom)
+			}
+		})
 	}
-	e := updater.NewSwarmExecutor(fake)
-
-	err := e.Execute(context.Background(), discovery.Job{ServiceID: "svc-myapp", Image: "registry.example.com/myapp:v2"})
-
-	require.Error(t, err)
-}
-
-func TestSwarmExecutor_RefusedJobIsNotExecuted(t *testing.T) {
-	fake := &fakeSwarmDockerClient{}
-	e := updater.NewSwarmExecutor(fake)
-	job := discovery.Job{ServiceID: "svc-myapp", Image: "x:v2", Refused: true, RefusedReason: "not opted in"}
-
-	err := e.Execute(context.Background(), job)
-
-	require.Error(t, err)
-	require.False(t, fake.inspectCalled)
-	require.False(t, fake.updateCalled)
 }