From aec21d23d9b731c12b4d48c5374510760b006780 Mon Sep 17 00:00:00 2001 From: Samuel Enocsson Date: Tue, 13 Oct 2026 12:06:49 +0200 Subject: [PATCH] fix: address review findings for self-update (#15) --- CLAUDE.md | 21 +++--- README.md | 10 ++- cmd/master/main.go | 2 +- .../specs/2026-10-12-master-agent-design.md | 18 +++-- internal/agent/inventory/collector.go | 68 +++++++++++++++---- internal/agent/inventory/collector_test.go | 59 +++++++++++++++- internal/agent/updater/compose.go | 53 +++++++++------ internal/agent/updater/compose_test.go | 31 +++++++-- internal/agent/updater/runner_test.go | 2 +- internal/agent/updater/selfupdate.go | 51 ++++++++++---- internal/agent/updater/selfupdate_test.go | 31 +++++---- internal/e2e/jobs_test.go | 4 +- internal/master/jobs/jobs.go | 26 ++++--- internal/master/jobs/jobs_test.go | 18 ++++- internal/master/model/job.go | 5 +- internal/master/poller/poller.go | 18 ++++- internal/master/poller/poller_test.go | 18 +++++ .../store/drift_merge_integration_test.go | 4 +- internal/master/store/merge.go | 2 +- internal/master/web/drift_test.go | 4 +- internal/master/web/server_test.go | 2 +- internal/master/web/versions_test.go | 2 +- internal/master/webhook/webhook.go | 12 ++++ internal/master/webhook/webhook_test.go | 15 ++++ internal/protocol/protocol.go | 15 ---- 25 files changed, 365 insertions(+), 126 deletions(-) diff --git a/CLAUDE.md b/CLAUDE.md index a09f305..e74eeab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -58,14 +58,12 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ the agent's own container whose source equals its destination.** Mounts are read per job (`ContainerInspect(SELF_ID)`) and checked after symlink resolution. This replaces the old `STACKS_ROOT`. -- **Self-update**: an agent's own Compose service is an ordinary target. Pull runs in the - old agent; `up -d` runs in a helper container made from the agent's image (same mounts), - labelled `se.shcizo.package-updater.self-update.*`. The new agent finds the helper at - startup (`updater.ResumeSelfUpdates`, before runner and connection start) and reports - the result under the old job ID. The old agent records no result once SIGTERM cancels - its context. Self jobs (`JobStart.Self`) run after every other queued job. An agent's - own Swarm service is refused (`ReasonSelfSwarm`). Needs protocol 4. The master updates - itself with no special code: the agent on its host resends the result. +- **Self-update** (an agent's own Compose service): only `up -d` runs in the helper + container; the new agent resumes the job (`updater.ResumeSelfUpdates`) before the runner + and connection start; the old agent records no result once SIGTERM cancels it. Jobs for + the agent's own service run last on the agent and the master starts them last in a + batch. An agent's own Swarm service is refused (`ReasonSelfSwarm`). Protocol 4 is needed, + also for any service with the `AGENT_IMAGE` repo. - **Unknown message types are logged and ignored; malformed messages close the connection.** - **Reconcile before routable**: the hub fails a reconnecting agent's jobs it does not list in `known_jobs`, and only then accepts jobs for that connection. @@ -131,8 +129,9 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ - **`protocol.Service.ConfigFiles` is the raw comma-joined label string**, so `Service` stays comparable with `==` (`inventory.Diff` depends on it). - **Helper containers carry `se.shcizo.package-updater.self-update.*` labels** and are - not inventoried (the collector skips any container with the `.job` label). The labels - are constants in `internal/protocol`. + not inventoried (the collector skips any container with the `.job` label). The label + strings are constants in `internal/agent/updater/selfupdate.go`, and the `.job` one is + repeated in `internal/agent/inventory`. - **Swarm "succeeded" means the daemon accepted the new spec**, not that tasks converged. - **Swarm updates need manager-node API access** for the agent doing the update. - **The image's default entrypoint is `updater-agent`.** The master overrides it with @@ -145,7 +144,7 @@ DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ knows about; only the waiting stops. The master never fails or cancels a timed-out job. - **A webhook job's target is `:`**, built from the service's own repo, never from the request's image string. With an explicit target the "follows no tag" - refusal does not apply; removed, built, container and Swarm self still do. + refusal does not apply; removed, built, container and an agent's own Swarm service still do. - **`available_*` is cleared by `FinishJob`** on success only when the job's `to_ref` (as created, without `@digest`) equals `available_ref`, `available_at <= job.created_at` (that version is now running) and, when the job's `to_ref` carries a digest, that digest diff --git a/README.md b/README.md index 1373a8f..00ca645 100644 --- a/README.md +++ b/README.md @@ -62,7 +62,7 @@ Content-Type: application/json in the request are ignored for matching; matching is case-sensitive). `tag` is optional. For every match the master records the version as *available* (the overview shows an "update available" badge), including services that cannot be updated automatically -(updater's own, `build:` and plain containers), so the badge shows for them too. Services +(`build:` services, plain containers and an agent's own Swarm service), so the badge shows for them too. Services in **manual** mode stop there. Services in **auto** mode also get an update job, and the request waits for those jobs to finish, at most `UPDATE_TIMEOUT`. @@ -299,7 +299,13 @@ any other service; the overview marks the agent's row "This agent". upgraded to a version with protocol 4 by hand once. - **Order.** A job for an agent's own service runs after every other queued job on that agent. A master and an agent on the same host share an image, so one push starts both - jobs, and the master's job has to finish before the agent replaces itself. + jobs, and the master's job has to finish before the agent replaces itself. The master + also starts a job for an agent's own service after the other jobs of the same poll + cycle or webhook request. Any other job that reaches an agent while it replaces itself + is marked lost when the new agent connects without it. +- **Docker socket.** The helper runs on the default network, so `DOCKER_HOST` only works + with a `docker.sock` bind mount. A TCP socket proxy that is reachable only on a Compose + network does not work for self-update. - **Swarm.** An agent's own Swarm service is refused: the agent cannot update the service it runs in. diff --git a/cmd/master/main.go b/cmd/master/main.go index e5349ff..36bfe90 100644 --- a/cmd/master/main.go +++ b/cmd/master/main.go @@ -104,7 +104,7 @@ func run() error { broker := events.NewBroker() agentHub := hub.New(st, broker, logger) - jobService := jobs.New(st, agentHub, broker, logger) + jobService := jobs.New(st, agentHub, broker, cfg.AgentImage, logger) provider, err := webauth.NewOIDCProvider(ctx, webauth.OIDCConfig{ Issuer: cfg.OIDCIssuer, ClientID: cfg.OIDCClientID, ClientSecret: cfg.OIDCClientSecret, diff --git a/docs/superpowers/specs/2026-10-12-master-agent-design.md b/docs/superpowers/specs/2026-10-12-master-agent-design.md index ad1c210..351ef02 100644 --- a/docs/superpowers/specs/2026-10-12-master-agent-design.md +++ b/docs/superpowers/specs/2026-10-12-master-agent-design.md @@ -229,7 +229,7 @@ Online/offline is held in memory only. | Master restarts mid-job | Jobs in `running` are reconciled when the agent reconnects; jobs the agent does not know are marked `failed` ("lost"). | | New trigger while a job for the same service is queued or running | No new job; only `available_*` is updated. Two jobs never race on one service. | | Job exceeds `UPDATE_TIMEOUT` | The agent cancels it, as today. | -| Agent updates itself | The old agent runs `up -d` in a helper container and waits. When the helper recreates the agent, the old agent is stopped and records nothing. The new agent finds the helper by its labels before it connects, so its hello lists the job; it reports the helper's outcome as the job's result. If the helper fails or the new agent never starts, the job stays `running` until the agent reconnects without it ("lost") or `UPDATE_TIMEOUT` ends the wait. | +| Agent updates itself | The old agent runs `up -d` in a helper container and waits. If the helper exits while the old agent still runs, the old agent reports the outcome. If the helper stops the old agent, the old agent records nothing; the new agent finds the helper by its labels before it connects, so its hello lists the job, and it reports the helper's outcome. If no new agent resumes the job, it is marked lost when the agent reconnects without it. If `UPDATE_TIMEOUT` passes first, the old agent reports a timeout and leaves the helper running; a later success is then not recorded. | | Polling fails for one image | Error stored in `last_poll_error` and shown; other images continue. | ### Self-update @@ -258,11 +258,17 @@ update itself"). - **Stopping.** When the agent is stopped (SIGTERM) while a job runs, the runner drops the job and records no result, so a "context canceled" result cannot overtake the new agent's. -- **Order.** `job.start` carries `self`; the agent runs a self job only when no other job - is queued (the master's job on a shared host has to finish first). Whether to use a - helper is decided by the agent from its own container ID, not from the flag. -- **Protocol 4.** The master routes a self job only to an agent speaking protocol 4 - (`agent too old to update itself`). +- **Order.** `job.start` carries `self`; the agent runs a job for its own service only + when no other job is queued (the master's job on a shared host has to finish first). + The master starts such jobs after the other jobs of the same poll cycle or webhook + request; messages to one agent arrive in order, so the others are queued first. A job + that reaches an agent while it replaces itself is marked lost when the new agent + connects without it. Whether to use a helper is decided by the agent from its own + container ID, not from the flag. +- **Protocol 4.** The master routes a job for an agent's own service, and any job for a + service with the `AGENT_IMAGE` repo (an older agent does not report itself), only to an + agent speaking protocol 4 (`agent too old to update itself`). The helper runs on the + default network, so `DOCKER_HOST` must be a `docker.sock` bind mount. ## Registry polling diff --git a/internal/agent/inventory/collector.go b/internal/agent/inventory/collector.go index cf7e878..6fff6aa 100644 --- a/internal/agent/inventory/collector.go +++ b/internal/agent/inventory/collector.go @@ -22,12 +22,14 @@ import ( ) const ( - labelProject = "com.docker.compose.project" - labelService = "com.docker.compose.service" - labelWorkingDir = "com.docker.compose.project.working_dir" - labelConfigFiles = "com.docker.compose.project.config_files" - labelEnvFile = "com.docker.compose.project.environment_file" - labelSwarmTask = "com.docker.swarm.service.id" + labelProject = "com.docker.compose.project" + labelService = "com.docker.compose.service" + labelWorkingDir = "com.docker.compose.project.working_dir" + labelConfigFiles = "com.docker.compose.project.config_files" + labelEnvFile = "com.docker.compose.project.environment_file" + labelSwarmTask = "com.docker.swarm.service.id" + labelSwarmName = "com.docker.swarm.service.name" + labelSelfUpdateJob = "se.shcizo.package-updater.self-update.job" ) // DockerClient is the subset of the Docker SDK the collector needs. @@ -61,6 +63,7 @@ type Collector struct { now func() time.Time mu sync.Mutex + self *selfInfo // the agent's own container, once inspected cache map[string]buildEntry // by Compose project warned map[string]bool // logged warnings: "" (check failed) and "/" (service not in config) } @@ -77,8 +80,15 @@ type buildEntry struct { // stackLabels are the Compose labels a build check needs. type stackLabels struct{ workingDir, configFiles, envFile string } -// New returns a Collector. selfID is a prefix of the agent's own container -// ID (inside a container, the hostname); that container is marked Self. +// selfInfo is what the inspect of the agent's own container tells. +type selfInfo struct { + id string // full container ID + swarmService string // name of the Swarm service the container is a task of, if any +} + +// New returns a Collector. selfID is the agent's own container: a prefix of +// its ID (inside a container, the hostname) or its name. That container is +// marked Self, and so is the Swarm service it is a task of. func New(cli DockerClient, selfID string) *Collector { return &Collector{cli: cli, selfID: selfID, now: time.Now, cache: map[string]buildEntry{}, warned: map[string]bool{}} } @@ -92,6 +102,29 @@ func (c *Collector) WithBuildCheck(check BuildCheck, log *slog.Logger) *Collecto return c } +// resolveSelf inspects the agent's own container once and remembers the +// answer. It returns nil while that fails (or selfID is empty), and the +// caller then compares container IDs with the selfID prefix. +func (c *Collector) resolveSelf(ctx context.Context) *selfInfo { + if c.selfID == "" { + return nil + } + c.mu.Lock() + defer c.mu.Unlock() + if c.self != nil { + return c.self + } + insp, err := c.cli.ContainerInspect(ctx, c.selfID) + if err != nil || insp.ContainerJSONBase == nil || insp.ID == "" { + return nil + } + c.self = &selfInfo{id: insp.ID} + if insp.Config != nil { + c.self.swarmService = insp.Config.Labels[labelSwarmName] + } + return c.self +} + // Hello returns the machine facts sent on connect. AgentVersion is left for // the caller to fill in. func (c *Collector) Hello(ctx context.Context) (protocol.Hello, error) { @@ -125,20 +158,21 @@ func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) { repoDigests[img.ID] = img.RepoDigests } + self := c.resolveSelf(ctx) byKey := make(map[string]protocol.Service) stacks := make(map[string]stackLabels) for _, ct := range containers { if _, isTask := ct.Labels[labelSwarmTask]; isTask { continue } - if _, isHelper := ct.Labels[protocol.LabelSelfUpdateJob]; isHelper { + if _, isHelper := ct.Labels[labelSelfUpdateJob]; isHelper { continue } ref, err := c.imageRef(ctx, ct) if err != nil { return nil, err } - svc := c.fromContainer(ct, ref, repoDigests[ct.ImageID]) + svc := c.fromContainer(ct, ref, repoDigests[ct.ImageID], self) if svc.Kind == protocol.KindCompose { if _, ok := stacks[svc.ComposeProject]; !ok { stacks[svc.ComposeProject] = stackLabels{svc.WorkingDir, svc.ConfigFiles, ct.Labels[labelEnvFile]} @@ -162,6 +196,7 @@ func (c *Collector) Snapshot(ctx context.Context) ([]protocol.Service, error) { continue } svc := fromSwarmService(s) + svc.Self = self != nil && self.swarmService != "" && svc.SwarmService == self.swarmService byKey[svc.Key()] = svc } } @@ -192,12 +227,12 @@ func (c *Collector) imageRef(ctx context.Context, ct container.Summary) (string, return insp.Config.Image, nil } -func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests []string) protocol.Service { +func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests []string, self *selfInfo) protocol.Service { svc := protocol.Service{ Image: ref, Digest: digestFor(ref, repoDigests), Status: containerStatus(ct.State), - Self: c.selfID != "" && strings.HasPrefix(ct.ID, c.selfID), + Self: c.isSelf(ct.ID, self), } project, service := ct.Labels[labelProject], ct.Labels[labelService] if project != "" && service != "" { @@ -216,6 +251,15 @@ func (c *Collector) fromContainer(ct container.Summary, ref string, repoDigests return svc } +// isSelf reports whether id is the agent's own container: by the inspected +// ID when there is one, else by the selfID prefix. +func (c *Collector) isSelf(id string, self *selfInfo) bool { + if self != nil { + return id == self.id + } + return c.selfID != "" && strings.HasPrefix(id, c.selfID) +} + func fromSwarmService(s swarm.Service) protocol.Service { ref := s.Spec.TaskTemplate.ContainerSpec.Image return protocol.Service{ diff --git a/internal/agent/inventory/collector_test.go b/internal/agent/inventory/collector_test.go index 24d5ee3..60e1878 100644 --- a/internal/agent/inventory/collector_test.go +++ b/internal/agent/inventory/collector_test.go @@ -138,7 +138,7 @@ func TestSnapshot_PlainContainerAndSwarmTaskSkipped(t *testing.T) { func TestSnapshot_SelfUpdateHelperSkipped(t *testing.T) { cli := &fakeDocker{containers: []container.Summary{ {ID: "h1", Names: []string{"/quirky_helper"}, Image: "sha256:agentimage", State: "running", - Labels: map[string]string{protocol.LabelSelfUpdateJob: "7"}}, + Labels: map[string]string{labelSelfUpdateJob: "7"}}, {ID: "p1", Names: []string{"/adhoc"}, Image: "redis:alpine", State: "exited"}, }} got, err := New(cli, "").Snapshot(context.Background()) @@ -158,6 +158,63 @@ func TestSnapshot_SelfFlag(t *testing.T) { assert.True(t, got[0].Self) } +func TestSnapshot_SelfFlagByContainerName(t *testing.T) { + cli := &fakeDocker{ + containers: []container.Summary{ + composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running"), + composeContainer("fedcba9876543210", "slusk", "web", "slusk:1", "sha256:j", "running"), + }, + inspect: map[string]container.InspectResponse{ + "updater-agent-agent-1": {ContainerJSONBase: &container.ContainerJSONBase{ID: "0123456789abcdef"}}, + }, + } + got, err := New(cli, "updater-agent-agent-1").Snapshot(context.Background()) + require.NoError(t, err) + require.Len(t, got, 2) + assert.False(t, got[0].Self, "slusk") + assert.True(t, got[1].Self, "updater-agent") +} + +func TestSnapshot_SelfFlagFallsBackToIDPrefixWhenInspectFails(t *testing.T) { + cli := &fakeDocker{ + containers: []container.Summary{composeContainer("0123456789abcdef", "updater-agent", "agent", "pu:1", "sha256:i", "running")}, + inspectErr: errors.New("boom"), + } + got, err := New(cli, "0123456789ab").Snapshot(context.Background()) + require.NoError(t, err) + require.Len(t, got, 1) + assert.True(t, got[0].Self) +} + +func TestSnapshot_AgentsOwnSwarmServiceIsSelf(t *testing.T) { + swarmSvc := func(name string) swarm.Service { + return swarm.Service{ + Spec: swarm.ServiceSpec{ + Annotations: swarm.Annotations{Name: name}, + TaskTemplate: swarm.TaskSpec{ContainerSpec: &swarm.ContainerSpec{Image: "pu:1"}}, + }, + ServiceStatus: &swarm.ServiceStatus{RunningTasks: 1, DesiredTasks: 1}, + } + } + cli := &fakeDocker{ + info: managerInfo(), + services: []swarm.Service{swarmSvc("pu_agent"), swarmSvc("pu_master")}, + inspect: map[string]container.InspectResponse{ + "abc123": { + ContainerJSONBase: &container.ContainerJSONBase{ID: "abc123full"}, + Config: &container.Config{Labels: map[string]string{labelSwarmName: "pu_agent"}}, + }, + }, + } + got, err := New(cli, "abc123").Snapshot(context.Background()) + require.NoError(t, err) + require.Len(t, got, 2) + assert.Equal(t, "pu_agent", got[0].SwarmService) + assert.True(t, got[0].Self) + assert.Equal(t, "pu_master", got[1].SwarmService) + assert.False(t, got[1].Self) +} + func TestSnapshot_SwarmServicesOnlyOnManager(t *testing.T) { svc := swarm.Service{ Spec: swarm.ServiceSpec{ diff --git a/internal/agent/updater/compose.go b/internal/agent/updater/compose.go index 9a47145..b7400b8 100644 --- a/internal/agent/updater/compose.go +++ b/internal/agent/updater/compose.go @@ -6,6 +6,7 @@ import ( "io" "os" "os/exec" + "slices" "strings" "time" @@ -98,10 +99,6 @@ func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run // When the service is the agent's own container, the recreate runs in a // helper container instead (see selfUpdate), because it stops this process. func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) (string, error) { - // What the job has printed so far goes into the helper's labels. - soFar := &tailBuffer{max: selfUpdateOutputMax} - out = io.MultiWriter(out, tailWriter{soFar}) - project, service := job.ComposeProject, job.ComposeService cts, err := e.cli.ContainerList(ctx, container.ListOptions{ All: true, @@ -116,14 +113,23 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) ( 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] - envFileLabel := cts[0].Labels[labelEnvFile] - self, err := inspectSelf(ctx, e.cli, e.selfID) if err != nil { return "", err } + isSelf := self.ContainerJSONBase != nil && self.ID != "" && + slices.ContainsFunc(cts, func(ct container.Summary) bool { return ct.ID == self.ID }) + // What an update of the agent's own service has printed so far goes into + // the helper's labels. + var soFar *tailBuffer + if isSelf { + soFar = &tailBuffer{max: selfUpdateOutputMax} + out = io.MultiWriter(out, tailWriter{soFar}) + } + workingDir := cts[0].Labels[labelWorkingDir] + configFiles := cts[0].Labels[labelConfigFiles] + envFileLabel := cts[0].Labels[labelEnvFile] + roots := rootsOf(self) files, dir, err := composeFiles(roots, configFiles) if err != nil { @@ -170,22 +176,29 @@ func (e *ComposeExecutor) Execute(ctx context.Context, job Job, out io.Writer) ( } base := composeBase(project, workingDir, files, envs) - isSelf := false - if self.ContainerJSONBase != nil && self.ID != "" { - for _, ct := range cts { - isSelf = isSelf || ct.ID == self.ID - } - } - for _, step := range [][]string{{"pull", service}, {"up", "-d", service}} { - args := append(append([]string(nil), base...), step...) - if isSelf && step[0] == "up" { - return e.selfUpdate(ctx, job, self, workingDir, args, pinned, out, soFar) - } + run := func(args []string) error { 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 composeError(args[len(base)], err, tail) } + return nil + } + if err := run(append(append([]string(nil), base...), "pull", service)); err != nil { + return "", err + } + up := append(append([]string(nil), base...), "up", "-d", service) + if isSelf { + return e.selfUpdate(ctx, job, self, workingDir, up, pinned, out, soFar) + } + if err := run(up); err != nil { + return "", err } return pinned, nil } + +// composeError describes a failed `docker compose ` call with the tail +// of its output. +func composeError(step string, err error, tail *tailBuffer) error { + return fmt.Errorf("docker compose %s: %w (output: %s)", step, err, strings.TrimSpace(tail.String())) +} diff --git a/internal/agent/updater/compose_test.go b/internal/agent/updater/compose_test.go index b7b387e..d2a4452 100644 --- a/internal/agent/updater/compose_test.go +++ b/internal/agent/updater/compose_test.go @@ -64,6 +64,7 @@ type fakeHelper struct { hold bool // ContainerWait never reports an exit logs string // what the helper printed onStart func() + state container.ContainerState // what an inspect reports; exited when empty created []createdHelper started []string @@ -79,6 +80,14 @@ func (h *fakeHelper) ContainerCreate(_ context.Context, cfg *container.Config, h return container.CreateResponse{ID: "helper1"}, nil } +func (h *fakeHelper) ContainerInspect(context.Context, string) (container.InspectResponse, error) { + state := h.state + if state == "" { + state = container.StateExited + } + return container.InspectResponse{ContainerJSONBase: &container.ContainerJSONBase{State: &container.State{Status: state}}}, nil +} + func (h *fakeHelper) ContainerStart(_ context.Context, id string, _ container.StartOptions) error { h.started = append(h.started, id) if h.onStart != nil { @@ -116,6 +125,14 @@ type fakeHelperCompose struct { h *fakeHelper } +// ContainerInspect answers for the agent's own container ("me") like fakeCompose +// and for any other ID like the helper. +func (f fakeHelperCompose) ContainerInspect(ctx context.Context, id string) (container.InspectResponse, error) { + if id == "me" { + return f.fakeCompose.ContainerInspect(ctx, id) + } + return f.h.ContainerInspect(ctx, id) +} func (f fakeHelperCompose) ContainerCreate(ctx context.Context, cfg *container.Config, host *container.HostConfig, n *network.NetworkingConfig, p *ocispec.Platform, name string) (container.CreateResponse, error) { return f.h.ContainerCreate(ctx, cfg, host, n, p, name) } @@ -528,13 +545,13 @@ func TestComposeSelfUpdate(t *testing.T) { assert.Equal(t, container.WaitConditionNextExit, c.helper.waitCond) labels := cfg.Labels assert.Len(t, labels, 6, "no compose labels") - assert.Equal(t, "7", labels[protocol.LabelSelfUpdateJob]) - assert.Equal(t, pinned, labels[protocol.LabelSelfUpdateRef]) - assert.Equal(t, "2026-10-13T09:00:00Z", labels[protocol.LabelSelfUpdateStarted]) - assert.Equal(t, "updater", labels[protocol.LabelSelfUpdateProject]) - assert.Equal(t, "agent", labels[protocol.LabelSelfUpdateService]) - assert.Contains(t, labels[protocol.LabelSelfUpdateOutput], "resolved "+pinned) - assert.Contains(t, labels[protocol.LabelSelfUpdateOutput], "$ docker "+strings.Join(withStep("pull", "agent"), " ")) + assert.Equal(t, "7", labels[labelSelfUpdateJob]) + assert.Equal(t, pinned, labels[labelSelfUpdateRef]) + assert.Equal(t, "2026-10-13T09:00:00Z", labels[labelSelfUpdateStarted]) + assert.Equal(t, "updater", labels[labelSelfUpdateProject]) + assert.Equal(t, "agent", labels[labelSelfUpdateService]) + assert.Contains(t, labels[labelSelfUpdateOutput], "resolved "+pinned) + assert.Contains(t, labels[labelSelfUpdateOutput], "$ docker "+strings.Join(withStep("pull", "agent"), " ")) }) } diff --git a/internal/agent/updater/runner_test.go b/internal/agent/updater/runner_test.go index 6e6eaca..b0951a5 100644 --- a/internal/agent/updater/runner_test.go +++ b/internal/agent/updater/runner_test.go @@ -260,7 +260,7 @@ func TestRunner_SelfJobsRunLast(t *testing.T) { assert.Equal(t, []string{"a:2", "a:4", "self:1", "self:3"}, order) } -func TestRunner_SelfJobRunsWhenAloneAndNonSelfSubmittedLater(t *testing.T) { +func TestRunner_SelfJobRunsWhenAlone(t *testing.T) { fe := newFakeExec() r := startRunner(t, fe, time.Minute) r.Submit("1", protocol.JobStart{Ref: "self:1", Self: true}) diff --git a/internal/agent/updater/selfupdate.go b/internal/agent/updater/selfupdate.go index 6f063c8..76fd91f 100644 --- a/internal/agent/updater/selfupdate.go +++ b/internal/agent/updater/selfupdate.go @@ -15,8 +15,17 @@ import ( "github.com/docker/docker/api/types/network" "github.com/docker/docker/pkg/stdcopy" ocispec "github.com/opencontainers/image-spec/specs-go/v1" +) - "github.com/shcizo/package-updater/internal/protocol" +// Labels on a self-update helper container. The new agent finds the helper +// by them, and the inventory skips containers that carry the job label. +const ( + labelSelfUpdateJob = "se.shcizo.package-updater.self-update.job" + labelSelfUpdateRef = "se.shcizo.package-updater.self-update.ref" + labelSelfUpdateStarted = "se.shcizo.package-updater.self-update.started" + labelSelfUpdateProject = "se.shcizo.package-updater.self-update.project" + labelSelfUpdateService = "se.shcizo.package-updater.self-update.service" + labelSelfUpdateOutput = "se.shcizo.package-updater.self-update.output" ) // selfUpdateOutputMax caps the job output stored in the helper container's @@ -32,6 +41,7 @@ const resumeSeparator = "--- agent restarted; output of the update helper ---\n" // HelperDocker is the subset of the Docker SDK the self-update helper // container needs. *client.Client satisfies it. type HelperDocker interface { + ContainerInspect(ctx context.Context, containerID string) (container.InspectResponse, error) ContainerCreate(ctx context.Context, config *container.Config, hostConfig *container.HostConfig, networkingConfig *network.NetworkingConfig, platform *ocispec.Platform, containerName string) (container.CreateResponse, error) ContainerStart(ctx context.Context, containerID string, options container.StartOptions) error ContainerWait(ctx context.Context, containerID string, condition container.WaitCondition) (<-chan container.WaitResponse, <-chan error) @@ -57,12 +67,12 @@ func (e *ComposeExecutor) selfUpdate(ctx context.Context, job Job, self containe WorkingDir: workingDir, Env: dockerEnv(), Labels: map[string]string{ - protocol.LabelSelfUpdateJob: job.ID, - protocol.LabelSelfUpdateRef: pinned, - protocol.LabelSelfUpdateStarted: job.Started.UTC().Format(time.RFC3339Nano), - protocol.LabelSelfUpdateProject: job.ComposeProject, - protocol.LabelSelfUpdateService: job.ComposeService, - protocol.LabelSelfUpdateOutput: outputLabel(soFar), + labelSelfUpdateJob: job.ID, + labelSelfUpdateRef: pinned, + labelSelfUpdateStarted: job.Started.UTC().Format(time.RFC3339Nano), + labelSelfUpdateProject: job.ComposeProject, + labelSelfUpdateService: job.ComposeService, + labelSelfUpdateOutput: outputLabel(soFar), }, } created, err := e.cli.ContainerCreate(ctx, cfg, &container.HostConfig{Mounts: helperMounts(self)}, nil, nil, "") @@ -72,7 +82,9 @@ func (e *ComposeExecutor) selfUpdate(ctx context.Context, job Job, self containe // Wait is registered before the start so a fast exit is not missed. waitCh, errCh := e.cli.ContainerWait(ctx, created.ID, container.WaitConditionNextExit) if err := e.cli.ContainerStart(ctx, created.ID, container.StartOptions{}); err != nil { - e.cli.ContainerRemove(context.WithoutCancel(ctx), created.ID, container.RemoveOptions{Force: true}) + if rmErr := e.cli.ContainerRemove(context.WithoutCancel(ctx), created.ID, container.RemoveOptions{Force: true}); rmErr != nil { + fmt.Fprintf(out, "warning: remove the update helper: %v\n", rmErr) + } return "", fmt.Errorf("start update helper: %w", err) } code, err := awaitExit(ctx, waitCh, errCh) @@ -154,7 +166,7 @@ func finishHelper(ctx context.Context, cli HelperDocker, id string, w io.Writer) // exitError describes a failed helper like a failed `docker compose` call. func exitError(code int64, tail *tailBuffer) error { - return fmt.Errorf("docker compose up: exit status %d (output: %s)", code, strings.TrimSpace(tail.String())) + return composeError("up", fmt.Errorf("exit status %d", code), tail) } // ResumeDocker is the subset of the Docker SDK ResumeSelfUpdates needs. @@ -186,16 +198,16 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run } cts, err := cli.ContainerList(ctx, container.ListOptions{ All: true, - Filters: filters.NewArgs(filters.Arg("label", protocol.LabelSelfUpdateJob)), + Filters: filters.NewArgs(filters.Arg("label", labelSelfUpdateJob)), }) if err != nil { return fmt.Errorf("list update helpers: %w", err) } for _, ct := range cts { - if ct.Labels[protocol.LabelSelfUpdateProject] != project || ct.Labels[protocol.LabelSelfUpdateService] != service { + if ct.Labels[labelSelfUpdateProject] != project || ct.Labels[labelSelfUpdateService] != service { continue } - jobID := ct.Labels[protocol.LabelSelfUpdateJob] + jobID := ct.Labels[labelSelfUpdateJob] if jobID == "" { log.Warn("self_update_helper_without_job", "helper", ct.ID) continue @@ -205,7 +217,7 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run started = time.Now() } log.Info("self_update_resumed", "job_id", jobID, "helper", ct.ID) - runner.Resume(jobID, started, resumeFunc(cli, ct.ID, ct.Labels[protocol.LabelSelfUpdateOutput], ref, labelErr)) + runner.Resume(jobID, started, resumeFunc(cli, ct.ID, ct.Labels[labelSelfUpdateOutput], ref, labelErr)) } return nil } @@ -214,11 +226,11 @@ func ResumeSelfUpdates(ctx context.Context, cli ResumeDocker, selfID string, run // labels. started is zero when it is missing or unreadable, and err then says // what is wrong. func parseHelperLabels(labels map[string]string) (ref string, started time.Time, err error) { - ref = labels[protocol.LabelSelfUpdateRef] + ref = labels[labelSelfUpdateRef] if ref == "" { err = errors.New("update helper has no ref label") } - raw := labels[protocol.LabelSelfUpdateStarted] + raw := labels[labelSelfUpdateStarted] if t, perr := time.Parse(time.RFC3339Nano, raw); perr == nil { started = t } else if err == nil { @@ -233,6 +245,15 @@ func parseHelperLabels(labels map[string]string) (ref string, started time.Time, // failed or its labels were unusable. func resumeFunc(cli HelperDocker, helperID, earlier, ref string, labelErr error) ResumeFunc { return func(ctx context.Context, out io.Writer) (string, error) { + // A helper that was created but never started would read as exited + // with code 0, so it must not be mistaken for a finished update. + if insp, err := cli.ContainerInspect(ctx, helperID); err == nil && insp.ContainerJSONBase != nil && + insp.State != nil && insp.State.Status == container.StateCreated { + if err := cli.ContainerRemove(context.WithoutCancel(ctx), helperID, container.RemoveOptions{Force: true}); err != nil { + fmt.Fprintf(out, "warning: remove the update helper: %v\n", err) + } + return "", errors.New("update helper was never started") + } waitCh, errCh := cli.ContainerWait(ctx, helperID, container.WaitConditionNotRunning) code, err := awaitExit(ctx, waitCh, errCh) if err != nil { diff --git a/internal/agent/updater/selfupdate_test.go b/internal/agent/updater/selfupdate_test.go index 103b013..31fca1f 100644 --- a/internal/agent/updater/selfupdate_test.go +++ b/internal/agent/updater/selfupdate_test.go @@ -22,12 +22,12 @@ func TestResumeSelfUpdates(t *testing.T) { helper := func(id, project, service string, extra map[string]string) container.Summary { labels := map[string]string{ - protocol.LabelSelfUpdateJob: "j-" + id, - protocol.LabelSelfUpdateRef: pinned, - protocol.LabelSelfUpdateStarted: started.Format(time.RFC3339Nano), - protocol.LabelSelfUpdateProject: project, - protocol.LabelSelfUpdateService: service, - protocol.LabelSelfUpdateOutput: "resolved " + pinned + "\n$ docker compose pull agent\n", + labelSelfUpdateJob: "j-" + id, + labelSelfUpdateRef: pinned, + labelSelfUpdateStarted: started.Format(time.RFC3339Nano), + labelSelfUpdateProject: project, + labelSelfUpdateService: service, + labelSelfUpdateOutput: "resolved " + pinned + "\n$ docker compose pull agent\n", } for k, v := range extra { if v == "" { @@ -44,6 +44,8 @@ func TestResumeSelfUpdates(t *testing.T) { name string helper container.Summary exit int64 + state container.ContainerState + noStarted bool // the old agent's start time is lost with the label wantStatus string wantErr string wantOutput string @@ -52,16 +54,19 @@ func TestResumeSelfUpdates(t *testing.T) { wantOutput: "resolved " + pinned + "\n$ docker compose pull agent\n" + separator + "Recreated\n"}, {name: "helper failed", helper: helper("h1", "updater", "agent", nil), exit: 1, wantStatus: protocol.ResultFailed, wantErr: "docker compose up: exit status 1 (output: Recreated)"}, - {name: "no ref label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateRef: ""}), + {name: "no ref label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateRef: ""}), wantStatus: protocol.ResultFailed, wantErr: "update helper has no ref label"}, - {name: "no started label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateStarted: ""}), + {name: "no started label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateStarted: ""}), + noStarted: true, wantStatus: protocol.ResultFailed, wantErr: `update helper has no usable started label ""`}, - {name: "no output label", helper: helper("h1", "updater", "agent", map[string]string{protocol.LabelSelfUpdateOutput: ""}), + {name: "helper never started", helper: helper("h1", "updater", "agent", nil), state: container.StateCreated, + wantStatus: protocol.ResultFailed, wantErr: "update helper was never started"}, + {name: "no output label", helper: helper("h1", "updater", "agent", map[string]string{labelSelfUpdateOutput: ""}), wantStatus: protocol.ResultSucceeded, wantOutput: separator + "Recreated\n"}, } for _, c := range cases { t.Run(c.name, func(t *testing.T) { - h := &fakeHelper{exit: c.exit, logs: "Recreated\n"} + h := &fakeHelper{exit: c.exit, state: c.state, logs: "Recreated\n"} unrelated := []container.Summary{ helper("other-project", "other", "agent", nil), helper("other-service", "updater", "web", nil), @@ -86,7 +91,9 @@ func TestResumeSelfUpdates(t *testing.T) { assert.Equal(t, "j-h1", got[0].ID) res := got[0].Result assert.Equal(t, c.wantStatus, res.Status) - assert.Equal(t, container.WaitConditionNotRunning, h.waitCond) + if c.state != container.StateCreated { + assert.Equal(t, container.WaitConditionNotRunning, h.waitCond) + } assert.Equal(t, []string{"h1"}, h.removed, "only the matching helper is touched") if c.wantStatus == protocol.ResultSucceeded { assert.Equal(t, pinned, res.Ref) @@ -94,7 +101,7 @@ func TestResumeSelfUpdates(t *testing.T) { } else { assert.Equal(t, c.wantErr, res.Error) } - if c.name != "no started label" { + if !c.noStarted { assert.True(t, started.Equal(res.StartedAt), "the old agent's start time is kept") } }) diff --git a/internal/e2e/jobs_test.go b/internal/e2e/jobs_test.go index 59b0a05..57da62c 100644 --- a/internal/e2e/jobs_test.go +++ b/internal/e2e/jobs_test.go @@ -119,7 +119,7 @@ func newJobEnv(t *testing.T) *jobEnv { m := startMaster(t, store) exec := newGateExecutor() env := &jobEnv{store: store, m: m, exec: exec, - jobs: jobs.New(store, m.hub, m.broker, slog.New(slog.NewTextHandler(io.Discard, nil))), + jobs: jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil))), agent: startJobAgent(t, m, "pua_ok", exec)} env.waitRoutable(t) return env @@ -282,7 +282,7 @@ func TestPhaseOneAgentGetsNoJobs(t *testing.T) { AgentID: ptr(agentID), ActualRef: slusk + ":1.4.2@sha256:old"} store.seedService(svcRow) m := startMaster(t, store) - js := jobs.New(store, m.hub, m.broker, slog.New(slog.NewTextHandler(io.Discard, nil))) + js := jobs.New(store, m.hub, m.broker, "gitea.shcizo.se/shcizo/package-updater:latest", slog.New(slog.NewTextHandler(io.Discard, nil))) // A raw client speaking the phase-1 protocol: its hello has no // protocol_version. diff --git a/internal/master/jobs/jobs.go b/internal/master/jobs/jobs.go index c7fecb8..26df766 100644 --- a/internal/master/jobs/jobs.go +++ b/internal/master/jobs/jobs.go @@ -40,16 +40,24 @@ type Publisher interface{ Publish(t events.Topic) } // Service starts update jobs. type Service struct { - store Store - hub Hub - pub Publisher - log *slog.Logger - now func() time.Time + store Store + hub Hub + pub Publisher + selfRepo string + log *slog.Logger + now func() time.Time } -// New returns a Service. -func New(store Store, hub Hub, pub Publisher, log *slog.Logger) *Service { - return &Service{store: store, hub: hub, pub: pub, log: log, now: time.Now} +// New returns a Service. agentImage is the master's AGENT_IMAGE. A service +// whose image repo equals its repo may be an agent that reports itself as +// another service (an agent older than protocol version 4 does not tell), so +// such a job is only sent to an agent that speaks protocol version +// protocol.MinSelfUpdateVersion. +func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service { + return &Service{ + store: store, hub: hub, pub: pub, log: log, now: time.Now, + selfRepo: imageref.Parse(agentImage).Repo, + } } // target splits a service's actual ref into the repo and tag to deploy. A bare @@ -245,7 +253,7 @@ func (s *Service) Start(ctx context.Context, svc model.Service, req StartRequest if p := imageref.Parse(ref); p.Digest != "" || p.Tag == "" { minProtocol = protocol.MinPinnedRefVersion } - if svc.Self { + if svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo { minProtocol = protocol.MinSelfUpdateVersion } r, routeErr := s.hub.Route(svc, minProtocol) diff --git a/internal/master/jobs/jobs_test.go b/internal/master/jobs/jobs_test.go index edcef17..b1d794d 100644 --- a/internal/master/jobs/jobs_test.go +++ b/internal/master/jobs/jobs_test.go @@ -16,6 +16,8 @@ import ( "github.com/shcizo/package-updater/internal/protocol" ) +const selfImage = "gitea.shcizo.se/shcizo/package-updater:latest" + type fakeStore struct { services map[int64]model.Service jobs map[int64]model.Job @@ -125,7 +127,7 @@ func ptr[T any](v T) *T { return &v } func newService(st *fakeStore, h *fakeHub) (*Service, *fakePub) { pub := &fakePub{} - return New(st, h, pub, slog.New(slog.NewTextHandler(io.Discard, nil))), pub + return New(st, h, pub, selfImage, slog.New(slog.NewTextHandler(io.Discard, nil))), pub } func TestUpdatable(t *testing.T) { @@ -417,6 +419,20 @@ func TestStart_SelfServiceNeedsASelfUpdatingAgent(t *testing.T) { require.Len(t, h.starts, 1) assert.True(t, h.starts[0].start.Self) }) + t.Run("a service with the agent image repo needs a self-updating agent but is not marked", func(t *testing.T) { + svc := composeSvc + svc.ActualRef = selfImage + st := &fakeStore{services: map[int64]model.Service{1: svc}, jobs: map[int64]model.Job{}} + h := &fakeHub{route: ok} + s, _ := newService(st, h) + + _, err := s.Start(context.Background(), svc, StartRequest{Trigger: model.TriggerManual}) + + require.NoError(t, err) + assert.Equal(t, []int{protocol.MinSelfUpdateVersion}, h.mins) + require.Len(t, h.starts, 1) + assert.False(t, h.starts[0].start.Self) + }) t.Run("a service that is not the agent's own is not marked", func(t *testing.T) { st := &fakeStore{services: map[int64]model.Service{1: composeSvc}, jobs: map[int64]model.Job{}} h := &fakeHub{route: ok} diff --git a/internal/master/model/job.go b/internal/master/model/job.go index 7bec9a1..e4ac160 100644 --- a/internal/master/model/job.go +++ b/internal/master/model/job.go @@ -49,7 +49,6 @@ func (e *NotUpdatableError) Error() string { return "not updatable: " + e.Reason // Reasons a service cannot be updated. const ( - ReasonSelf = "updated manually outside the system" ReasonContainer = "plain containers cannot be updated" ReasonNoTag = "follows no tag" ReasonRemoved = "service no longer exists" @@ -70,14 +69,14 @@ type MergeRefusedError struct { func (e *MergeRefusedError) Error() string { return "cannot merge: " + e.Reason } -// Reasons two services cannot be merged. ReasonSelf is reused for the -// updater itself. +// Reasons two services cannot be merged. const ( ReasonMergeSame = "a service cannot be merged into itself" ReasonMergeLiveOld = "the old service still exists" ReasonMergeGoneNew = "the new service no longer exists" ReasonMergeKind = "only services of the same kind can be merged" ReasonMergeContainer = "plain containers cannot be merged" + ReasonMergeSelf = "an agent's own service cannot be merged" ) // JobStatus is the state of an update job. diff --git a/internal/master/poller/poller.go b/internal/master/poller/poller.go index eb1b17b..7bdbb95 100644 --- a/internal/master/poller/poller.go +++ b/internal/master/poller/poller.go @@ -7,6 +7,7 @@ import ( "context" "errors" "log/slog" + "slices" "strings" "sync" "time" @@ -209,14 +210,29 @@ func (p *Poller) record(ctx context.Context, l lookup, svcs []model.Service, sta } stats.ok++ p.setPollError(ctx, svcs, "", stats, func(s model.Service) bool { return s.LastPollError != "" }) + var auto []model.Service for _, svc := range svcs { if !p.suggest(ctx, svc, l, stats) { continue } if autoWanted(svc, l.ref, l.digest) { - p.startAuto(ctx, svc.ID, l.ref, l.digest, stats) + auto = append(auto, svc) } } + // An agent's own service goes last: jobs that reach an agent while it + // replaces itself are lost, and messages to one agent arrive in order. + slices.SortStableFunc(auto, func(a, b model.Service) int { return selfRank(a) - selfRank(b) }) + for _, svc := range auto { + p.startAuto(ctx, svc.ID, l.ref, l.digest, stats) + } +} + +// selfRank sorts an agent's own service after all others. +func selfRank(svc model.Service) int { + if svc.Self { + return 1 + } + return 0 } // suggest marks l.digest as the available version of svc unless the service diff --git a/internal/master/poller/poller_test.go b/internal/master/poller/poller_test.go index 531575e..139e27d 100644 --- a/internal/master/poller/poller_test.go +++ b/internal/master/poller/poller_test.go @@ -297,6 +297,24 @@ func TestNewDigest(t *testing.T) { } } +func TestAgentsOwnServiceIsStartedLast(t *testing.T) { + ref := repoA + ":1.4" + own := svc(1, ref+"@sha256:old") + own.UpdateMode, own.Self = model.ModeAuto, true + other := svc(2, ref+"@sha256:old") + other.UpdateMode = model.ModeAuto + st := &fakeStore{services: []model.Service{own, other}} + reg := &fakeRegistry{digests: map[string]string{ref: "sha256:new"}} + j := &fakeJobs{} + p, _ := newPoller(st, j, reg, Config{}) + + p.Cycle(context.Background()) + + require.Len(t, j.starts, 2) + assert.Equal(t, int64(2), j.starts[0].svcID) + assert.Equal(t, int64(1), j.starts[1].svcID) +} + func TestSameSuggestionIsIdempotent(t *testing.T) { ref := repoA + ":1.4" running := svc(1, ref+"@sha256:old") diff --git a/internal/master/store/drift_merge_integration_test.go b/internal/master/store/drift_merge_integration_test.go index 1433920..c051986 100644 --- a/internal/master/store/drift_merge_integration_test.go +++ b/internal/master/store/drift_merge_integration_test.go @@ -418,11 +418,11 @@ func TestMergeService_Refusals(t *testing.T) { {"the updater itself (new)", func(t *testing.T, s *Store, f moveFixture) error { mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.to) return s.MergeService(context.Background(), f.from, f.to) - }, refused(model.ReasonSelf)}, + }, refused(model.ReasonMergeSelf)}, {"the updater itself (old)", func(t *testing.T, s *Store, f moveFixture) error { mustExec(t, s, `UPDATE services SET self = true WHERE id = $1`, f.from) return s.MergeService(context.Background(), f.from, f.to) - }, refused(model.ReasonSelf)}, + }, refused(model.ReasonMergeSelf)}, {"active job on the old service", func(t *testing.T, s *Store, f moveFixture) error { _, err := s.CreateJob(context.Background(), newJob(f.from, f.a)) require.NoError(t, err) diff --git a/internal/master/store/merge.go b/internal/master/store/merge.go index 0a4f0a7..9c22f6b 100644 --- a/internal/master/store/merge.go +++ b/internal/master/store/merge.go @@ -243,7 +243,7 @@ func mergeRefusal(from, to mergeRow) string { case from.kind == string(protocol.KindContainer): return model.ReasonMergeContainer case from.self || to.self: - return model.ReasonSelf + return model.ReasonMergeSelf } return "" } diff --git a/internal/master/web/drift_test.go b/internal/master/web/drift_test.go index 108d638..fa3b7c0 100644 --- a/internal/master/web/drift_test.go +++ b/internal/master/web/drift_test.go @@ -273,8 +273,8 @@ func TestServiceDetail_DriftAndMoveData(t *testing.T) { {name: "swarm service can merge, no suggestion", id: 3, wantSources: sources}, {name: "container cannot merge", id: 4}, {name: "self cannot merge", id: 5}, - {name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonSelf}, - wantBlocked: model.ReasonSelf, wantFrom: &move, wantSources: sources}, + {name: "restore refusal is passed on", id: 1, restorable: &model.NotUpdatableError{Reason: model.ReasonContainer}, + wantBlocked: model.ReasonContainer, wantFrom: &move, wantSources: sources}, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { diff --git a/internal/master/web/server_test.go b/internal/master/web/server_test.go index c487a83..4e9f069 100644 --- a/internal/master/web/server_test.go +++ b/internal/master/web/server_test.go @@ -539,7 +539,7 @@ func TestServicePage_ShowsUpdateButton(t *testing.T) { } func TestServicePage_DisabledReason(t *testing.T) { - for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonNoTag, model.ReasonRemoved, model.ReasonBuilt} { + for _, reason := range []string{model.ReasonContainer, model.ReasonNoTag, model.ReasonRemoved, model.ReasonBuilt} { t.Run(reason, func(t *testing.T) { srv, _, _, _, jobs := newTestEnv(t) jobs.updatable = &model.NotUpdatableError{Reason: reason} diff --git a/internal/master/web/versions_test.go b/internal/master/web/versions_test.go index 0b29796..e5628cf 100644 --- a/internal/master/web/versions_test.go +++ b/internal/master/web/versions_test.go @@ -68,7 +68,7 @@ func TestServicePage_VersionPickerWarnsForAutoServices(t *testing.T) { } func TestServicePage_VersionPickerHiddenWhenServiceCannotBeUpdated(t *testing.T) { - for _, reason := range []string{model.ReasonSelf, model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} { + for _, reason := range []string{model.ReasonContainer, model.ReasonRemoved, model.ReasonBuilt} { t.Run(reason, func(t *testing.T) { srv, store, _, _, jobs := newTestEnv(t) store.services[0].ImageRepo = sluskRepo diff --git a/internal/master/webhook/webhook.go b/internal/master/webhook/webhook.go index 36ebfbf..cd8fe0e 100644 --- a/internal/master/webhook/webhook.go +++ b/internal/master/webhook/webhook.go @@ -15,6 +15,7 @@ import ( "log/slog" "net/http" "regexp" + "slices" "strconv" "strings" "time" @@ -181,6 +182,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { resp := response{RequestID: requestID, Image: req.Image, Tag: req.Tag, Matched: len(services)} pending := map[int]int64{} // index into resp.Results -> job ID + // An agent's own service goes last: jobs that reach an agent while it + // replaces itself are lost, and messages to one agent arrive in order. + slices.SortStableFunc(services, func(a, b model.Service) int { return selfRank(a) - selfRank(b) }) for _, svc := range services { res, job := h.handle(work, svc, req.Tag, started, log) resp.Results = append(resp.Results, res) @@ -252,6 +256,14 @@ func (h *Handler) handle(ctx context.Context, svc model.Service, tag string, sta return res, nil } +// selfRank sorts an agent's own service after all others. +func selfRank(svc model.Service) int { + if svc.Self { + return 1 + } + return 0 +} + func (h *Handler) setJob(res *result, job model.Job) { res.JobID = job.ID res.JobURL = h.cfg.PublicURL + "/jobs/" + strconv.FormatInt(job.ID, 10) diff --git a/internal/master/webhook/webhook_test.go b/internal/master/webhook/webhook_test.go index a39eba7..faff3e7 100644 --- a/internal/master/webhook/webhook_test.go +++ b/internal/master/webhook/webhook_test.go @@ -500,6 +500,21 @@ func TestMixedAutoAndManual(t *testing.T) { assert.Len(t, e.store.avail, 3) } +func TestAgentsOwnServiceIsStartedLast(t *testing.T) { + own := compose(1, model.ModeAuto, slusk+":1") + own.Self = true + e := newEnv(t, own, compose(2, model.ModeAuto, slusk+":1")) + e.jobs.jobFor[1] = model.Job{Status: model.JobSucceeded} + e.jobs.jobFor[2] = model.Job{Status: model.JobSucceeded} + + e.post(`{"image":"` + slusk + `","tag":"2"}`) + + calls := e.jobs.startCalls() + require.Len(t, calls, 2) + assert.Equal(t, int64(2), calls[0].svc.ID) + assert.Equal(t, int64(1), calls[1].svc.ID) +} + func TestClientCancelStopsWaitingButNotStarting(t *testing.T) { e := newEnv(t, compose(1, model.ModeAuto, slusk+":1.4.2")) e.h.cfg.Timeout = time.Minute diff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go index aa0cd80..fec63bc 100644 --- a/internal/protocol/protocol.go +++ b/internal/protocol/protocol.go @@ -31,21 +31,6 @@ const MinPinnedRefVersion = 3 // update in a helper container nor resume it after restarting. const MinSelfUpdateVersion = 4 -// SelfUpdateLabelPrefix starts the names of the labels on the helper -// container that finishes an agent's self-update. The new agent finds the -// helper by them, and the inventory skips containers that carry them. -const SelfUpdateLabelPrefix = "se.shcizo.package-updater.self-update." - -// Labels on a self-update helper container. -const ( - LabelSelfUpdateJob = SelfUpdateLabelPrefix + "job" - LabelSelfUpdateRef = SelfUpdateLabelPrefix + "ref" - LabelSelfUpdateStarted = SelfUpdateLabelPrefix + "started" - LabelSelfUpdateProject = SelfUpdateLabelPrefix + "project" - LabelSelfUpdateService = SelfUpdateLabelPrefix + "service" - LabelSelfUpdateOutput = SelfUpdateLabelPrefix + "output" -) - // MaxJobOutput caps the job output an agent keeps and the master stores, in // bytes. Only the tail is kept. With JSON escaping (worst case 6 bytes per // byte) a JobResult carrying this much output still fits the 1 MiB read limit.