1390 lines
86 KiB
Markdown
1390 lines
86 KiB
Markdown
# Master/Agent Phase 2 — Jobs and Manual Update Implementation Plan
|
||
|
||
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. Code in "Frozen contracts" is final; code inside tasks is given only where it is subtle — elsewhere the task gives the contract and the test cases, and the implementer writes the code.
|
||
|
||
**Goal:** A person clicks *Update* on a service in the master UI; the owning agent re-pulls the service's current tag (resolved to `repo:tag@digest`) and redeploys it, streaming its log live to the browser. Job history is kept in Postgres. `cmd/server` and the old single-instance updater code are removed.
|
||
|
||
**Architecture:** The master creates an `update_jobs` row and sends `job.start` over the agent's existing WebSocket. The agent queues it in a single-worker FIFO runner, resolves the tag's digest through the Docker daemon, and deploys: Compose by writing an agent-owned `compose.override.yaml` and running `docker compose pull` + `up -d`; Swarm by `ServiceUpdate` with the pinned image. Output flows back as `job.progress`, the outcome as `job.result`, which the agent keeps until the master answers `job.ack`. On every hello the agent lists the job IDs it holds; the master marks its own queued/running jobs for that agent that are not listed as failed ("lost"). The browser re-fetches a job's full log from Postgres on each SSE signal, so dropped signals are harmless.
|
||
|
||
**Tech Stack:** unchanged from phase 1, plus `gopkg.in/yaml.v3` and `github.com/distribution/reference` promoted from indirect to direct. `github.com/prometheus/*` disappears with `internal/metrics`.
|
||
|
||
**Spec:** `docs/superpowers/specs/2026-10-12-master-agent-design.md`, phase 2. Phase 1 plan: `docs/superpowers/plans/2026-10-12-master-agent-phase-1-inventory.md` (its conventions still apply).
|
||
|
||
## Global Constraints
|
||
|
||
- Go `1.26.3` in `go.mod`; Dockerfile stays on `golang:1.26-alpine`.
|
||
- `go test ./...` passes with no Docker daemon and no Postgres. Postgres tests carry `//go:build integration` and need `DATABASE_URL`. Run the race detector on every package that gets goroutines in this phase: `go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/`.
|
||
- Interfaces are declared in the consuming package. Hand-written fakes, `stretchr/testify`, table-driven where there is more than one case.
|
||
- Every exported identifier gets a GoDoc comment. Conventional Commits; every commit message ends with ` (#7)`.
|
||
- UI text is English. Every state-changing UI request is a POST carrying the CSRF token.
|
||
- WebSocket read limit stays `1 << 20` on both sides. Every websocket write goes through one writer goroutine per connection (both sides) — no other goroutine calls `wsjson.Write`.
|
||
- Never edit operator files: the agent writes exactly one file, `<stack dir>/compose.override.yaml`, and only if it is absent or starts with the header line.
|
||
- Agent tokens and `AGENT_TOKEN` values never appear in logs (unchanged).
|
||
- **One shared worktree, one Go module, implementers in parallel.** While W2–W4 run, each implementer runs `go vet`/`go test` only on the packages they own (listed per workstream below), because another workstream's package may be mid-edit. Stage only your own paths with `git add <paths>` — never `git add -A` or `git add .`. If `git` reports `index.lock` exists, wait a few seconds and retry; never delete the lock file. Full `go vet ./... && go test ./...` runs in W5 and in the final verification.
|
||
- The agent may only touch Compose stacks in directories that are bind-mounted into its own container read-write **at the same path as on the host**. There is no `STACKS_ROOT` setting.
|
||
|
||
## Deliberate deviations from the spec in this phase
|
||
|
||
- **Only "re-pull the current tag" exists.** The job's target is always the service's current `repo:tag`; the agent resolves the digest. No version picker (phase 4), no webhook (phase 3). Services without a tag (`repo@digest`, "follows no tag") cannot be updated; the UI says why.
|
||
- **`cmd/server`, `internal/api`, `internal/config`, `internal/discovery`, `internal/selfupdate` and `internal/metrics` are deleted.** There is no `POST /update` anywhere until phase 3 (see PR notes). `internal/selfupdate` does not compile without `discovery.Job`, and the problem it solves (flushing an HTTP response before recreating yourself) does not exist in the agent model; git history keeps it as the starting point the spec mentions. `internal/metrics` was only used by the old queue and server.
|
||
- **`internal/updater` moves to `internal/agent/updater`** (`git mv`, history kept). Its blocking `Queue.Submit` (built for the synchronous webhook) is replaced by `Runner` (async, outbox, live output). The executors are rewritten against `protocol.JobStart`. `discovery.IsInsideRoot` moves into the same package; image matching is dropped (`imageref.Parse(x).Repo` is equivalent to `NormaliseImage`).
|
||
- **Override file: one name, and refuse when any other override exists.** Verified empirically against Docker Compose v5.1.2: Compose auto-loads `compose.override.{yml,yaml}` / `docker-compose.override.{yml,yaml}` next to *any* default base file, but loads **only one** (first of `compose.override.yml`, `compose.override.yaml`, `docker-compose.override.yml`, `docker-compose.override.yaml`) and warns about the rest. So the agent writes `compose.override.yaml` and refuses the job when any of the other three exists — writing ours next to an operator's `docker-compose.override.yml` would silently disable theirs on the next manual `docker compose up`. Stacks started with explicit `-f` files do not load the override on a manual `up`; the agent's own jobs always pass it explicitly.
|
||
- **The override file is never pruned.** If a service is deleted from the stack file, its `image:` entry stays in `compose.override.yaml` and a manual `docker compose up -d` would recreate it. Documented in the README; pruning needs the base file parsed and is deferred.
|
||
- **`STACKS_ROOT` is replaced by the agent's own mounts.** The spec keeps "the `STACKS_ROOT` path check on the agent"; this phase keeps the check but changes where the allowed directories come from. The agent inspects its own container (`SELF_ID`, default the hostname) and treats every bind mount that is read-write and has `Source == Destination` as an allowed root. Every config file and the override path must be inside one of them (`filepath.Rel` check, also after resolving symlinks). One less setting to keep in sync with the volume list, and a read-only or path-translated mount — which could not work anyway, since labels hold host paths — is refused with a clear message. Mounts are read on every Compose job (one `ContainerInspect`), not cached: it costs one local API call per job and never acts on stale data. An agent not running in a container, or whose `SELF_ID` does not resolve, has no roots, and Compose jobs fail; the agent logs a warning at startup. On Docker Desktop/OrbStack, bind sources are translated host paths and may not equal the destination; that is a development-only limitation.
|
||
- **`job.start` carries no paths.** The agent reads `working_dir` and `config_files` from its own Docker daemon (container labels) at execution time. The inventory still reports `ConfigFiles` (shown on the service page), but a stale or forged path from the master never reaches `docker compose`.
|
||
- **`protocol.Service.ConfigFiles` is the raw comma-joined label string**, not a slice, so `Service` stays comparable with `==` (`inventory.Diff` depends on it).
|
||
- **Swarm "succeeded" means the daemon accepted the new spec**, not that tasks converged. The inventory shows `degraded` until they do.
|
||
- **Job timestamps:** `created_at` is the master's clock; `started_at`/`finished_at` come from the agent's `JobResult` (the master sets `started_at` from its own clock on the first `job.progress` only if the result never arrives).
|
||
- **Output cap 128 KiB (tail)**, not 256 KiB: a `JobResult` carrying the whole output must fit the 1 MiB read limit even with worst-case JSON escaping (6×).
|
||
- **SSE topic for one job is `job-<id>`** (not `job:<id>`), so it is unambiguous inside `hx-trigger="sse:job-42"`.
|
||
- **Revoking an agent fails its queued/running jobs** ("agent revoked"). Not in the spec, but without it a job on a machine that never comes back would block that service forever (one active job per service).
|
||
- **A job whose agent never reconnects stays `running`** until the agent is revoked. The master has no job timeout of its own; the agent's `UPDATE_TIMEOUT` applies on the agent.
|
||
- **Registry auth supports inline `auths` entries only** (`auth` or `identitytoken` in `config.json`). `credsStore`/`credHelpers` are not available inside the agent container; the agent logs a warning at startup if the file names them.
|
||
- **Self-protection is enforced on the master only.** The agent does not re-check whether a job targets itself.
|
||
- **Unknown message types are logged and ignored on both sides** (phase 1 closed the connection). Unparseable messages still close it.
|
||
|
||
## Review Focus
|
||
|
||
| Failure case | Test that covers it |
|
||
|---|---|
|
||
| A job finishes while the agent is disconnected: the result is resent after reconnect and recorded exactly once; a result resent after a lost `job.ack` is acked again but not re-applied. | e2e `TestJobFinishedWhileDisconnectedIsRecordedOnce` (W5-1); store `TestFinishJob_OnlyOnceAndOnlyByOwner` (W3-1) |
|
||
| Master restart mid-job: the agent still has the job → it is **not** marked lost and its result lands. Agent restart mid-job: the job is marked "lost". | e2e `TestMasterRestartKeepsJobTheAgentKnows`, `TestAgentRestartMarksJobLost` (W5-1); hub `TestHub_ReconcilesBeforeRoutable` (W3-2) |
|
||
| A job is never sent over a connection that replaced the one routed to (agent reconnected in between); the job fails "agent offline" instead of running after being marked lost. | hub `TestHub_StartJobOnReplacedConnectionFails` (W3-2) |
|
||
| An agent cannot write progress to or finish another agent's job (leaked token). | store `TestJobProgress_OwnerOnly`, `TestFinishJob_OnlyOnceAndOnlyByOwner` (W3-1) |
|
||
| Two clicks on Update at once create one job; the second gets the first back. | store `TestCreateJob_OneActivePerService` (concurrent, W3-1); jobs `TestTrigger` "active job" case (W3-3) |
|
||
| The master refuses jobs for itself and the agents (same image repo as `AGENT_IMAGE`) and for rows flagged `self`. | jobs `TestUpdatable` (W3-3); web `TestServicePage_DisabledReason` (W4-1) |
|
||
| The agent never overwrites an operator's override file, and never writes next to another auto-loaded override name. | updater `TestWriteOverride` (W2-4) |
|
||
| Every config file and the override path are inside a read-write, same-path bind mount of the agent's own container, also after resolving symlinks; read-only and path-translated mounts are refused. | updater `TestMountRoots`, `TestComposeFiles` (W2-4) |
|
||
| Concurrent writers on one websocket (job progress, acks, inventory) are serialised. | `go test -race` on conn and hub tests that start a job while inventory flows (W2-6, W3-2) |
|
||
| A phase-1 agent (protocol 0) never receives `job.start`; the job fails "agent too old for updates". | e2e `TestPhaseOneAgentGetsNoJobs` (W5-1); hub `TestRoute` (W3-2) |
|
||
| A Swarm job routed to an agent whose daemon is in a different cluster is refused before any update. | updater `TestSwarmExecutor` "cluster mismatch" (W2-5) |
|
||
| A job running past `UPDATE_TIMEOUT` is cancelled and fails with a timeout error. | updater `TestRunner_Timeout` (W2-3) |
|
||
| A 128 KiB output of escape characters still arrives in one `job.result`. | hub `TestHub_LargeResultArrives` (W3-2) |
|
||
| A message type the peer does not know is ignored without dropping the connection. | hub `TestHub_IgnoresUnknownType` (W3-2); conn `TestAgent_IgnoresUnknownType` (W2-6) |
|
||
|
||
## PR notes (put this in the PR description)
|
||
|
||
> **Production risk — read before merging or tagging a release.** This PR removes `cmd/server` and changes the image's default `ENTRYPOINT` to `updater-agent`. The currently running single-instance updater (`docker-compose.example.yml` style: `image: …/package-updater:latest`, default entrypoint, and its own `se.shcizo.auto-update=true` label) will, the next time anything pulls `:latest` for it — its own `/update` webhook, Watchtower or Shepherd — restart as an agent without `MASTER_URL` and exit in a crash loop. Its `/update` endpoint, which the Gitea action calls, disappears with it.
|
||
> Before the next `v*` tag after this merges, the operator must either pin the old updater's image to the last pre-phase-2 tag (and remove its opt-in label), or not release until phase 3 has moved `POST /update` to the master. Agents and the master must be on this release together: a phase-1 agent connects fine but every job sent to it fails "agent too old for updates".
|
||
|
||
## Verified facts this plan relies on
|
||
|
||
- **Compose override loading** (Docker Compose v5.1.2, tested in a temp dir with `docker compose config` and a throwaway `docker compose create`): see the deviation above. The `com.docker.compose.project.config_files` label lists every file used, as absolute paths joined by `,`, **including an auto-loaded override**. The agent therefore removes `<dir>/compose.override.yaml` from the label's list and appends it once, last.
|
||
- **`ServiceUpdate` by name:** the daemon's `getService` (`daemon/cluster/helpers.go`) resolves full ID, then exact name, then ID prefix. The executor still inspects by name, checks `Spec.Name`, and updates by the returned `ID` with the returned `Version`, so it updates exactly the object it inspected.
|
||
- **Digest resolution:** `client.DistributionInspect(ctx, "repo:tag", encodedAuth)` (API ≥ 1.30) returns `Descriptor.Digest`; the daemon contacts the registry, so it works for any registry the daemon can pull from.
|
||
- **Registry auth encoding:** `registry.EncodeAuthConfig(registry.AuthConfig{...})` (`api/types/registry`).
|
||
|
||
## Workstreams and file ownership
|
||
|
||
W1 lands first (one small commit set). W2, W3, W4 then run in parallel in the same worktree. W5 runs after W2–W4 have committed everything.
|
||
|
||
| Workstream | Owner | Owns (only this workstream edits these) |
|
||
|---|---|---|
|
||
| W1 contracts | first implementer | `internal/protocol/protocol.go`, `protocol_test.go`; `internal/master/model/model.go`, new `model/job.go`, `model_test.go`; `internal/master/events/broker.go`, `broker_test.go`; new `internal/master/store/migrations/0002_jobs.sql` |
|
||
| W2 agent | implementer-agent | `internal/agent/**` (incl. new `internal/agent/updater/`), `cmd/agent/`, deletions of `cmd/server`, `internal/api`, `internal/config`, `internal/discovery`, `internal/updater`, `internal/selfupdate`, `internal/metrics`; `Dockerfile`; `docker-compose.example.yml`, `docker-compose.local.yml`, `env.sample` (deleted); new `deploy/agent.compose.yml`; `go.mod`, `go.sum` |
|
||
| W3 master | implementer-master | `internal/master/store/**` (except the migration file W1 adds), `internal/master/hub/**`, new `internal/master/jobs/`, `internal/master/config/`, `cmd/master/` |
|
||
| W4 web | implementer-web | `internal/master/web/**` (Go, templates, `static/app.css`, tests), `internal/master/webauth/auth.go` + `auth_test.go` (one helper only) |
|
||
| W5 integration | implementer-master | `internal/e2e/**`, `CLAUDE.md`, `README.md`, `deploy/master.stack.yml` |
|
||
|
||
Shared touch points, resolved by contract rather than co-editing:
|
||
- `cmd/master/main.go` (W3) calls `web.New` with W4's new signature (C7). W4's **first** commit (Task W4-0) is that signature change plus the matching one-line change to the call in `cmd/master` (the only line W4 touches there), so the module builds at every commit. W3 does W3-4 only after W4-0 is committed.
|
||
- `go.mod`/`go.sum`: only W2 runs `go mod tidy`. W3/W4 add no dependencies.
|
||
- `internal/master/model` is W1's. If W3 or W4 needs a model change, they ask the W1 owner (lead) rather than editing it.
|
||
|
||
---
|
||
|
||
## Frozen contracts
|
||
|
||
These are final. W2–W4 code against them; W1 commits them verbatim.
|
||
|
||
### C1. `internal/protocol/protocol.go` (complete file)
|
||
|
||
```go
|
||
// Package protocol defines the messages agents and the master exchange over
|
||
// their WebSocket connection. Every message is an Envelope whose Payload is
|
||
// one of the payload types below, selected by Type.
|
||
//
|
||
// Both sides close the connection on a message they cannot parse (bad JSON,
|
||
// or a known Type whose payload does not decode). A message with an unknown
|
||
// Type is logged and ignored, so a newer peer can add message types without
|
||
// breaking an older one.
|
||
package protocol
|
||
|
||
import (
|
||
"encoding/json"
|
||
"fmt"
|
||
"time"
|
||
)
|
||
|
||
// Version is the protocol version this build speaks. Agents send it in
|
||
// Hello. Phase-1 agents do not send the field, which reads as 0.
|
||
const Version = 2
|
||
|
||
// MinJobVersion is the lowest agent protocol version the master sends jobs to.
|
||
const MinJobVersion = 2
|
||
|
||
// 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.
|
||
const MaxJobOutput = 128 << 10
|
||
|
||
// Message types sent from agent to master.
|
||
const (
|
||
TypeHello = "hello"
|
||
TypeSnapshot = "snapshot"
|
||
TypeUpsert = "upsert"
|
||
TypeRemove = "remove"
|
||
TypeJobProgress = "job.progress"
|
||
TypeJobResult = "job.result"
|
||
)
|
||
|
||
// Message types sent from master to agent.
|
||
const (
|
||
TypeJobStart = "job.start"
|
||
TypeJobAck = "job.ack"
|
||
)
|
||
|
||
// Envelope wraps every message on the wire. ID is the job ID (decimal) on
|
||
// job.* messages and empty on inventory messages.
|
||
type Envelope struct {
|
||
Type string `json:"type"`
|
||
ID string `json:"id,omitempty"`
|
||
Payload json.RawMessage `json:"payload,omitempty"`
|
||
}
|
||
|
||
// Hello is the first message an agent sends on every connection.
|
||
type Hello struct {
|
||
AgentVersion string `json:"agent_version"`
|
||
ProtocolVersion int `json:"protocol_version,omitempty"`
|
||
Hostname string `json:"hostname"`
|
||
SwarmManager bool `json:"swarm_manager"`
|
||
SwarmClusterID string `json:"swarm_cluster_id,omitempty"`
|
||
// KnownJobs lists every job ID the agent holds: queued, running, or
|
||
// finished but not yet acknowledged. The master fails its queued and
|
||
// running jobs for this agent that are not listed ("lost").
|
||
KnownJobs []string `json:"known_jobs,omitempty"`
|
||
}
|
||
|
||
// Kind says how a service is deployed.
|
||
type Kind string
|
||
|
||
// Service kinds.
|
||
const (
|
||
KindCompose Kind = "compose"
|
||
KindSwarm Kind = "swarm"
|
||
KindContainer Kind = "container"
|
||
)
|
||
|
||
// Service is one deployable unit as an agent sees it. Which location fields
|
||
// are set depends on Kind.
|
||
type Service struct {
|
||
Kind Kind `json:"kind"`
|
||
ComposeProject string `json:"compose_project,omitempty"`
|
||
ComposeService string `json:"compose_service,omitempty"`
|
||
WorkingDir string `json:"working_dir,omitempty"`
|
||
// ConfigFiles is the raw com.docker.compose.project.config_files label:
|
||
// absolute host paths joined by ",". Kept as a string so Service stays
|
||
// comparable with ==.
|
||
ConfigFiles string `json:"config_files,omitempty"`
|
||
SwarmService string `json:"swarm_service,omitempty"`
|
||
ContainerName string `json:"container_name,omitempty"`
|
||
Image string `json:"image"`
|
||
Digest string `json:"digest,omitempty"`
|
||
Status string `json:"status"`
|
||
Self bool `json:"self,omitempty"`
|
||
}
|
||
|
||
// Key identifies the service within one agent's report. It is stable across
|
||
// restarts and is what Remove refers to.
|
||
func (s Service) Key() string {
|
||
switch s.Kind {
|
||
case KindCompose:
|
||
return "compose|" + s.ComposeProject + "|" + s.ComposeService
|
||
case KindSwarm:
|
||
return "swarm|" + s.SwarmService
|
||
default:
|
||
return "container|" + s.ContainerName
|
||
}
|
||
}
|
||
|
||
// Snapshot is the agent's complete inventory. Sent after every Hello.
|
||
type Snapshot struct {
|
||
Services []Service `json:"services"`
|
||
}
|
||
|
||
// Upsert reports a service that appeared or changed.
|
||
type Upsert struct {
|
||
Service Service `json:"service"`
|
||
}
|
||
|
||
// Remove reports that the service with Key is gone.
|
||
type Remove struct {
|
||
Key string `json:"key"`
|
||
}
|
||
|
||
// JobStart asks the agent to deploy Ref to one service. Ref is "repo:tag";
|
||
// the agent resolves the tag's current digest and deploys "repo:tag@digest".
|
||
// The agent reads the Compose working directory and config files from its
|
||
// own Docker daemon, not from this message.
|
||
type JobStart struct {
|
||
Kind Kind `json:"kind"`
|
||
ComposeProject string `json:"compose_project,omitempty"`
|
||
ComposeService string `json:"compose_service,omitempty"`
|
||
SwarmClusterID string `json:"swarm_cluster_id,omitempty"`
|
||
SwarmService string `json:"swarm_service,omitempty"`
|
||
Ref string `json:"ref"`
|
||
}
|
||
|
||
// JobProgress carries output produced since the previous JobProgress for the
|
||
// same job. The first JobProgress of a job (Output may be empty) tells the
|
||
// master the job is running.
|
||
type JobProgress struct {
|
||
Output string `json:"output"`
|
||
}
|
||
|
||
// Job result statuses.
|
||
const (
|
||
ResultSucceeded = "succeeded"
|
||
ResultFailed = "failed"
|
||
)
|
||
|
||
// JobResult is a job's outcome. The agent keeps it until the master sends
|
||
// JobAck for the same ID, and resends it after every reconnect until then.
|
||
type JobResult struct {
|
||
Status string `json:"status"` // ResultSucceeded or ResultFailed
|
||
Ref string `json:"ref,omitempty"` // deployed "repo:tag@digest", set on success
|
||
Error string `json:"error,omitempty"`
|
||
// Output is the job's complete output, tail-capped at MaxJobOutput. It
|
||
// replaces whatever the master assembled from JobProgress messages.
|
||
Output string `json:"output"`
|
||
OutputTruncated bool `json:"output_truncated,omitempty"`
|
||
StartedAt time.Time `json:"started_at"`
|
||
FinishedAt time.Time `json:"finished_at"`
|
||
}
|
||
|
||
// JobAck tells the agent the master has stored the JobResult with the same
|
||
// ID. It has no fields.
|
||
type JobAck struct{}
|
||
|
||
// Encode wraps payload in an Envelope of the given type.
|
||
func Encode(typ string, payload any) (Envelope, error) {
|
||
raw, err := json.Marshal(payload)
|
||
if err != nil {
|
||
return Envelope{}, fmt.Errorf("encode %s: %w", typ, err)
|
||
}
|
||
return Envelope{Type: typ, Payload: raw}, nil
|
||
}
|
||
|
||
// EncodeJob wraps a job message's payload in an Envelope carrying the job ID.
|
||
func EncodeJob(typ, jobID string, payload any) (Envelope, error) {
|
||
env, err := Encode(typ, payload)
|
||
if err != nil {
|
||
return Envelope{}, err
|
||
}
|
||
env.ID = jobID
|
||
return env, nil
|
||
}
|
||
|
||
// Decode unpacks env's payload into T.
|
||
func Decode[T any](env Envelope) (T, error) {
|
||
var v T
|
||
if err := json.Unmarshal(env.Payload, &v); err != nil {
|
||
return v, fmt.Errorf("decode %s: %w", env.Type, err)
|
||
}
|
||
return v, nil
|
||
}
|
||
```
|
||
|
||
Wire rules both sides follow:
|
||
|
||
| Message | Direction | `ID` | Receiver on bad payload | Receiver on unknown/foreign job ID |
|
||
|---|---|---|---|---|
|
||
| `job.start` | M→A | job ID | close | (n/a) duplicate ID ignored |
|
||
| `job.ack` | M→A | job ID | close | ignore |
|
||
| `job.progress` | A→M | job ID | close | ignore (non-integer ID: log + ignore) |
|
||
| `job.result` | A→M | job ID | close | **ack anyway** (non-integer ID too), so the agent drops it; log `job_result_ignored` |
|
||
| anything else unknown | both | — | — | log at warn, ignore, keep the connection |
|
||
|
||
Agent send order on each connection: `hello` (with `ProtocolVersion: protocol.Version`, `KnownJobs`) → `snapshot` → every unacknowledged `job.result` → then live traffic. A store error while the master handles a job message closes the connection (as in phase 1), so the agent reconnects and resends.
|
||
|
||
### C2. `internal/master/model` additions
|
||
|
||
`model.go` — add three fields to `Service` (W1):
|
||
|
||
```go
|
||
WorkingDir string
|
||
ConfigFiles string // raw config_files label, paths joined by ","
|
||
SwarmClusterID string
|
||
...
|
||
DesiredRef string // set by the last successful job; "" if never updated
|
||
ActualRef string
|
||
...
|
||
UpdatedAt time.Time
|
||
RemovedAt *time.Time
|
||
```
|
||
|
||
New file `job.go` (complete):
|
||
|
||
```go
|
||
package model
|
||
|
||
import (
|
||
"errors"
|
||
"time"
|
||
)
|
||
|
||
// Sentinel errors for jobs.
|
||
var (
|
||
// ErrJobActive means the service already has a queued or running job.
|
||
// CreateJob and jobs.Service.Trigger return it together with that job.
|
||
ErrJobActive = errors.New("an update is already in progress for this service")
|
||
// ErrAgentOffline means no connected agent can run the job.
|
||
ErrAgentOffline = errors.New("agent offline")
|
||
// ErrAgentTooOld means the only agent that could run the job speaks a
|
||
// protocol version without jobs.
|
||
ErrAgentTooOld = errors.New("agent too old for updates")
|
||
)
|
||
|
||
// NotUpdatableError says why a service cannot be updated. Reason is shown
|
||
// in the UI as is.
|
||
type NotUpdatableError struct {
|
||
Reason string
|
||
}
|
||
|
||
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"
|
||
)
|
||
|
||
// JobStatus is the state of an update job.
|
||
type JobStatus string
|
||
|
||
// Job statuses. A job moves queued -> running -> succeeded|failed, or
|
||
// straight from queued to failed.
|
||
const (
|
||
JobQueued JobStatus = "queued"
|
||
JobRunning JobStatus = "running"
|
||
JobSucceeded JobStatus = "succeeded"
|
||
JobFailed JobStatus = "failed"
|
||
)
|
||
|
||
// JobTrigger says what created a job. Only TriggerManual is used in phase 2.
|
||
type JobTrigger string
|
||
|
||
// Job triggers.
|
||
const (
|
||
TriggerManual JobTrigger = "manual"
|
||
TriggerWebhook JobTrigger = "webhook"
|
||
TriggerPoll JobTrigger = "poll"
|
||
)
|
||
|
||
// Job is one update of one service.
|
||
type Job struct {
|
||
ID int64
|
||
ServiceID int64
|
||
ServiceName string // Service.DisplayName() of the job's service
|
||
AgentID *int64 // agent the job was sent to; nil if it never was
|
||
AgentName string
|
||
FromRef string
|
||
ToRef string // "repo:tag" when created; "repo:tag@digest" after success
|
||
Trigger JobTrigger
|
||
RequestedBy string // OIDC subject for manual jobs
|
||
Status JobStatus
|
||
Output string // empty in ListJobs results
|
||
OutputTruncated bool
|
||
Error string
|
||
CreatedAt time.Time
|
||
StartedAt *time.Time
|
||
FinishedAt *time.Time
|
||
}
|
||
|
||
// Active reports whether the job is queued or running.
|
||
func (j Job) Active() bool { return j.Status == JobQueued || j.Status == JobRunning }
|
||
|
||
// NewJob is what CreateJob needs to insert a queued job.
|
||
type NewJob struct {
|
||
ServiceID int64
|
||
AgentID *int64
|
||
FromRef string
|
||
ToRef string
|
||
Trigger JobTrigger
|
||
RequestedBy string
|
||
CreatedAt time.Time
|
||
}
|
||
|
||
// JobFinish is a job's outcome as reported by the agent.
|
||
type JobFinish struct {
|
||
Status JobStatus // JobSucceeded or JobFailed
|
||
Ref string // deployed "repo:tag@digest"; set on success
|
||
Error string
|
||
Output string
|
||
OutputTruncated bool
|
||
StartedAt time.Time // agent clock
|
||
FinishedAt time.Time // agent clock
|
||
}
|
||
|
||
// JobFilter selects jobs for ListJobs. ServiceID 0 means all services.
|
||
// Limit 0 means 100.
|
||
type JobFilter struct {
|
||
ServiceID int64
|
||
Limit int
|
||
}
|
||
|
||
// Route is the hub's choice of agent connection for a job. ConnID identifies
|
||
// the connection, so a job is never sent over a connection that replaced the
|
||
// one chosen.
|
||
type Route struct {
|
||
AgentID int64
|
||
ConnID uint64
|
||
}
|
||
```
|
||
|
||
### C3. `internal/master/events/broker.go` additions
|
||
|
||
```go
|
||
import (
|
||
"strconv"
|
||
"sync"
|
||
)
|
||
|
||
// Topics.
|
||
const (
|
||
Services Topic = "services"
|
||
Agents Topic = "agents"
|
||
Jobs Topic = "jobs" // a job was created or changed status
|
||
)
|
||
|
||
// JobTopic is signalled when job id gets new output or changes status. The
|
||
// name avoids ":" so it is safe inside hx-trigger="sse:job-42".
|
||
func JobTopic(id int64) Topic { return Topic("job-" + strconv.FormatInt(id, 10)) }
|
||
```
|
||
|
||
Who publishes what (W3 implements, W4 listens):
|
||
|
||
| Event | Topics |
|
||
|---|---|
|
||
| job created (any outcome) | `Jobs`, `JobTopic(id)` |
|
||
| `job.progress` applied | `Jobs`, `JobTopic(id)` |
|
||
| job finished / failed / lost / revoked | `Jobs`, `JobTopic(id)`; plus `Services` on success (desired ref changed) |
|
||
|
||
`web.events` needs no change: it already writes `event: <topic>`.
|
||
|
||
### C4. Migration `internal/master/store/migrations/0002_jobs.sql` (complete)
|
||
|
||
```sql
|
||
ALTER TABLE services ADD COLUMN config_files text NOT NULL DEFAULT '';
|
||
ALTER TABLE services ADD COLUMN desired_ref text NOT NULL DEFAULT '';
|
||
|
||
CREATE TABLE update_jobs (
|
||
id bigserial PRIMARY KEY,
|
||
service_id bigint NOT NULL REFERENCES services (id),
|
||
agent_id bigint REFERENCES agents (id),
|
||
from_ref text NOT NULL DEFAULT '',
|
||
to_ref text NOT NULL,
|
||
trigger text NOT NULL CHECK (trigger IN ('manual', 'webhook', 'poll')),
|
||
requested_by text NOT NULL DEFAULT '',
|
||
status text NOT NULL CHECK (status IN ('queued', 'running', 'succeeded', 'failed')),
|
||
output text NOT NULL DEFAULT '',
|
||
output_truncated boolean NOT NULL DEFAULT false,
|
||
error text NOT NULL DEFAULT '',
|
||
created_at timestamptz NOT NULL,
|
||
started_at timestamptz,
|
||
finished_at timestamptz
|
||
);
|
||
|
||
-- One queued or running job per service. CreateJob relies on this index to
|
||
-- turn a second concurrent click into ErrJobActive.
|
||
CREATE UNIQUE INDEX update_jobs_one_active ON update_jobs (service_id)
|
||
WHERE status IN ('queued', 'running');
|
||
CREATE INDEX update_jobs_by_service ON update_jobs (service_id, id DESC);
|
||
CREATE INDEX update_jobs_active_by_agent ON update_jobs (agent_id)
|
||
WHERE status IN ('queued', 'running');
|
||
```
|
||
|
||
### C5. Store methods (W3 implements on `*store.Store`; W3/W4 consumers declare the subsets they use)
|
||
|
||
```go
|
||
// GetService returns one service, including a removed one (RemovedAt set).
|
||
// model.ErrNotFound if the ID does not exist.
|
||
GetService(ctx context.Context, id int64) (model.Service, error)
|
||
|
||
// CreateJob inserts a queued job. If the service already has a queued or
|
||
// running job, it returns that job and model.ErrJobActive (detected via the
|
||
// unique index, so it holds under concurrent calls).
|
||
CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
|
||
|
||
// FailJob sets a queued or running job to failed with msg and finished_at =
|
||
// at. Returns false (no error) if the job is no longer active.
|
||
FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
|
||
|
||
// GetJob returns one job with its output. model.ErrNotFound if missing.
|
||
GetJob(ctx context.Context, id int64) (model.Job, error)
|
||
|
||
// ListJobs returns jobs newest first, without Output.
|
||
ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error)
|
||
|
||
// FailLostJobs fails every queued or running job sent to agentID whose ID is
|
||
// not in known, with error "lost: the agent did not know this job when it
|
||
// reconnected". Returns the IDs it failed.
|
||
FailLostJobs(ctx context.Context, agentID int64, known []int64, at time.Time) ([]int64, error)
|
||
|
||
// JobProgress appends chunk to the output of a queued or running job sent to
|
||
// agentID, keeping the last protocol.MaxJobOutput characters, and moves a
|
||
// queued job to running (started_at = at if unset). Returns false if no
|
||
// such active job belongs to agentID.
|
||
JobProgress(ctx context.Context, agentID, jobID int64, chunk string, at time.Time) (bool, error)
|
||
|
||
// FinishJob records the agent's result for a queued or running job sent to
|
||
// agentID: status, error, output (replacing what progress assembled),
|
||
// started_at (kept if already set), finished_at; to_ref = f.Ref when set.
|
||
// On success it also sets the service's desired_ref = f.Ref, in the same
|
||
// transaction. Returns false if no such active job belongs to agentID.
|
||
FinishJob(ctx context.Context, agentID, jobID int64, f model.JobFinish) (bool, error)
|
||
```
|
||
|
||
Changed existing methods (W3): `upsertService`/`ApplySnapshot`/`UpsertService` also write `config_files`; `ListServices` also returns `ConfigFiles` and `DesiredRef`; `RevokeAgent` also fails the agent's queued/running jobs with "agent revoked" in its transaction.
|
||
|
||
### C6. `internal/master/jobs` (W3) — consumed by web (W4)
|
||
|
||
```go
|
||
// Package jobs decides whether a service may be updated and starts update
|
||
// jobs on the right agent.
|
||
package jobs
|
||
|
||
// Store is the storage jobs needs. *store.Store satisfies it.
|
||
type Store interface {
|
||
GetService(ctx context.Context, id int64) (model.Service, error)
|
||
CreateJob(ctx context.Context, j model.NewJob) (model.Job, error)
|
||
FailJob(ctx context.Context, id int64, msg string, at time.Time) (bool, error)
|
||
GetJob(ctx context.Context, id int64) (model.Job, error)
|
||
}
|
||
|
||
// Hub routes jobs to agent connections. *hub.Hub satisfies it.
|
||
type Hub interface {
|
||
// Route picks the connection for svc: model.ErrAgentOffline or
|
||
// model.ErrAgentTooOld when there is none.
|
||
Route(svc model.Service) (model.Route, error)
|
||
// StartJob sends job.start over exactly the routed connection.
|
||
// model.ErrAgentOffline if that connection is gone or was replaced.
|
||
StartJob(r model.Route, jobID int64, start protocol.JobStart) error
|
||
}
|
||
|
||
// Publisher receives change signals. *events.Broker satisfies it.
|
||
type Publisher interface{ Publish(t events.Topic) }
|
||
|
||
// New returns a Service. agentImage is the master's AGENT_IMAGE; services
|
||
// whose image repo equals its repo are the updater itself and are refused.
|
||
func New(store Store, hub Hub, pub Publisher, agentImage string, log *slog.Logger) *Service
|
||
|
||
// Updatable returns nil or a *model.NotUpdatableError. It looks only at the
|
||
// service row, so the UI can call it for every page render.
|
||
func (s *Service) Updatable(svc model.Service) error
|
||
|
||
// Trigger starts a manual update of the service to its current tag. It
|
||
// returns the job, which may already be failed ("agent offline", "agent too
|
||
// old for updates"); a *model.NotUpdatableError when the service cannot be
|
||
// updated (no job created); the active job and model.ErrJobActive when one
|
||
// is queued or running; model.ErrNotFound for an unknown service.
|
||
func (s *Service) Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
|
||
```
|
||
|
||
### C7. `internal/master/web` constructor (W4) — called by `cmd/master` (W3)
|
||
|
||
```go
|
||
// Jobs starts updates and explains why a service cannot be updated.
|
||
// *jobs.Service satisfies it.
|
||
type Jobs interface {
|
||
Updatable(svc model.Service) error
|
||
Trigger(ctx context.Context, serviceID int64, requestedBy string) (model.Job, error)
|
||
}
|
||
|
||
// Store gains (in addition to phase 1's methods):
|
||
GetService(ctx context.Context, id int64) (model.Service, error)
|
||
GetJob(ctx context.Context, id int64) (model.Job, error)
|
||
ListJobs(ctx context.Context, f model.JobFilter) ([]model.Job, error)
|
||
|
||
func New(store Store, hub Hub, jobs Jobs, broker Broker, cfg Config, log *slog.Logger) (*Server, error)
|
||
```
|
||
|
||
`cmd/master` wiring (W3):
|
||
|
||
```go
|
||
agentHub := hub.New(st, broker, logger)
|
||
jobService := jobs.New(st, agentHub, broker, cfg.AgentImage, logger)
|
||
ui, err := web.New(st, agentHub, jobService, broker, web.Config{...}, logger)
|
||
```
|
||
|
||
### C8. Agent configuration and public agent APIs (W2; used by W4 snippet and W5 e2e)
|
||
|
||
Environment (added to `internal/agent/config.Config`):
|
||
|
||
| Variable | Field | Default | Validation |
|
||
|---|---|---|---|
|
||
| `UPDATE_TIMEOUT` | `UpdateTimeout time.Duration` | `5m` | `time.ParseDuration`, must be > 0 |
|
||
| `DOCKER_CONFIG` | `DockerConfig string` | `""` (anonymous registry access) | if set: `$DOCKER_CONFIG/config.json` must exist and parse, else fail fast. Same variable `docker compose` reads, so `pull` uses the same credentials. |
|
||
|
||
There is no `STACKS_ROOT`. The allowed stack directories are the agent container's bind mounts that are read-write with `Source == Destination` (read on every Compose job via `ContainerInspect(SELF_ID)`). Each stacks directory must be mounted as `/path:/path` (no `:ro`), because Compose labels hold host paths.
|
||
|
||
```go
|
||
// package conn
|
||
type Jobs interface {
|
||
Submit(id string, start protocol.JobStart)
|
||
Ack(id string)
|
||
Known() []string
|
||
TakeProgress() []updater.Progress
|
||
Pending() []updater.Finished
|
||
Changed() <-chan struct{}
|
||
}
|
||
// Config gains ProgressInterval time.Duration // default 500ms
|
||
func New(cfg Config, src Source, jobs Jobs, log *slog.Logger) *Agent
|
||
|
||
// package updater
|
||
type Executor interface {
|
||
Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error)
|
||
}
|
||
type Progress struct{ ID, Output string }
|
||
type Finished struct {
|
||
ID string
|
||
Result protocol.JobResult
|
||
}
|
||
func NewRunner(exec Executor, timeout time.Duration) *Runner // *Runner satisfies conn.Jobs
|
||
func (r *Runner) Run(ctx context.Context)
|
||
```
|
||
|
||
---
|
||
|
||
## W1 — Contracts
|
||
|
||
### Task W1-1: Protocol, model, events, migration
|
||
|
||
**Files:**
|
||
- Modify: `internal/protocol/protocol.go`, `internal/protocol/protocol_test.go`
|
||
- Modify: `internal/master/model/model.go`, `internal/master/model/model_test.go`; Create: `internal/master/model/job.go`
|
||
- Modify: `internal/master/events/broker.go`, `internal/master/events/broker_test.go`
|
||
- Create: `internal/master/store/migrations/0002_jobs.sql`
|
||
|
||
- [ ] **Step 1: Write the failing tests**
|
||
- `protocol_test.go`: `TestEncodeJob_CarriesID` — `EncodeJob(TypeJobStart, "42", JobStart{Kind: KindSwarm, SwarmService: "gitea_gitea", Ref: "gitea/gitea:1.27"})` round-trips through `json.Marshal`/`Unmarshal` + `Decode[JobStart]` with `ID == "42"`; `TestHello_PhaseOneDecodesAsVersionZero` — `{"agent_version":"x","hostname":"h","swarm_manager":false}` decodes to `ProtocolVersion 0`, `KnownJobs nil`; `TestEnvelope_InventoryHasNoID` — `Encode(TypeSnapshot, …)` marshals without an `"id"` key.
|
||
- `model_test.go`: `TestJob_Active` table over the four statuses.
|
||
- `broker_test.go`: `TestJobTopic` — `JobTopic(42) == "job-42"`.
|
||
- [ ] **Step 2:** Run `go test ./internal/protocol ./internal/master/model ./internal/master/events` — FAIL (undefined identifiers).
|
||
- [ ] **Step 3:** Apply C1, C2, C3 exactly; add C4's file.
|
||
- [ ] **Step 4:** Run `go build ./... && go test ./...` — PASS. (The rest of the tree must still compile: `ConfigFiles` is a string, so `inventory.Diff`'s `!=` still works.)
|
||
- [ ] **Step 5: Commit** — `feat(protocol): job messages, protocol version and job model (#7)`
|
||
|
||
---
|
||
|
||
## W2 — Agent
|
||
|
||
### Task W2-1: Remove the old updater and move what survives
|
||
|
||
**Files:**
|
||
- `git mv internal/updater internal/agent/updater`; then delete `internal/agent/updater/{queue.go,queue_test.go,worker.go,worker_test.go,executor.go}`. Keep `compose_executor.go`, `swarm_executor.go`, `swarm_executor_test.go` for W2-4/W2-5 to rewrite (history follows the rename).
|
||
- `git mv internal/discovery/pathcheck.go internal/agent/updater/pathcheck.go` and `pathcheck_test.go` likewise; change `package discovery` → `package updater`.
|
||
- Delete: `cmd/server/`, `internal/api/`, `internal/config/`, the rest of `internal/discovery/`, `internal/selfupdate/`, `internal/metrics/`.
|
||
|
||
- [ ] **Step 1:** Do the moves and deletions. The moved executors no longer compile (they reference `discovery.Job`) until W2-4/W2-5 rewrite them, so W2-1, W2-3, W2-4 and W2-5 land as **one commit** once `go build ./...` is green. Never commit a tree that does not build.
|
||
- [ ] **Step 2:** Promote the two dependencies with `go get gopkg.in/yaml.v3 github.com/distribution/reference`. Run `go mod tidy` only when `go build ./...` is green for the whole module (other workstreams may be mid-edit); if it is not, leave tidy to W5. Confirm `github.com/prometheus/*` left `go.mod` and `gopkg.in/yaml.v3`, `github.com/distribution/reference` are direct.
|
||
- [ ] **Step 3:** `grep -rn "internal/\(api\|config\|discovery\|selfupdate\|metrics\|updater\)\"" --include='*.go' .` returns nothing.
|
||
- [ ] **Step 4: Commit** (with W2-3..W2-5) — `refactor(agent)!: move updater under the agent and remove cmd/server (#7)` with a `BREAKING CHANGE:` footer naming the removed `/update` endpoint and the new default entrypoint.
|
||
|
||
Packages W2 tests while W2–W4 run in parallel: `go vet ./internal/agent/... ./cmd/agent/ && go test -race ./internal/agent/...`.
|
||
|
||
### Task W2-2: Agent config and inventory `ConfigFiles`
|
||
|
||
**Files:** `internal/agent/config/config.go`, `config_test.go`; `internal/agent/inventory/collector.go`, `collector_test.go`
|
||
|
||
- [ ] **Step 1: Failing tests**
|
||
- `TestLoad_JobSettings` (table): defaults (`UpdateTimeout 5m`, `DockerConfig ""`); `UPDATE_TIMEOUT=90s` → 90s; `UPDATE_TIMEOUT=0`/`abc` → error; `DOCKER_CONFIG=<dir without config.json>` → error; with a valid file → ok.
|
||
- Collector: a Compose container with label `com.docker.compose.project.config_files=/s/a/compose.yaml,/s/a/compose.override.yaml` yields `ConfigFiles` equal to the raw label value.
|
||
- [ ] **Step 2:** Run `go test ./internal/agent/config ./internal/agent/inventory` — FAIL.
|
||
- [ ] **Step 3:** Implement. `config.Load` only validates `DOCKER_CONFIG`'s file exists and is JSON; parsing into credentials is W2-5's `LoadRegistryAuth`.
|
||
- [ ] **Step 4:** PASS. **Commit:** `feat(agent): UPDATE_TIMEOUT and DOCKER_CONFIG settings (#7)`
|
||
|
||
### Task W2-3: Job runner (queue, outbox, timeout)
|
||
|
||
**Files:** Create `internal/agent/updater/runner.go`, `runner_test.go`
|
||
|
||
**Contract:** C8 `Executor`, `Progress`, `Finished`, `NewRunner`, plus `Submit`, `Changed`, `Known`, `TakeProgress`, `Pending`, `Ack`, `Run`.
|
||
|
||
- [ ] **Step 1: Failing tests** (fake executor that writes `"pulling <ref>\n"` to `out` and blocks on a channel; fails for refs starting with `bad`):
|
||
- `TestRunner_FIFOSingleWorker`: submit 1, 2; executor records max concurrency = 1; results in `Pending()` in finishing order 1, 2; 1 succeeded with `Ref "a:1@sha256:aa"` and `Output "pulling a:1\n"`; 2 failed with the executor's error.
|
||
- `TestRunner_DuplicateSubmitIgnored`: second `Submit("1", …)` with a different ref does not run.
|
||
- `TestRunner_KnownAndAck`: `Known()` lists queued, running and finished-unacked IDs (sorted); after `Ack("1")` it is gone; `Ack` of a running job is ignored.
|
||
- `TestRunner_TakeProgress`: a started job appears once with `""` output even before writing; output written between two calls appears once; a finished job no longer appears.
|
||
- `TestRunner_Timeout`: timeout 20ms, executor blocks on ctx → result failed, `Error == "timed out after 20ms"`.
|
||
- `TestRunner_OutputCapped`: executor writes `protocol.MaxJobOutput + 10` bytes → `Result.Output` is the last `MaxJobOutput` bytes, `OutputTruncated` true.
|
||
- Run all with `-race`.
|
||
- [ ] **Step 2:** `go test -race ./internal/agent/updater -run Runner` — FAIL.
|
||
- [ ] **Step 3: Implement.** The state handling is subtle; use this shape (prototyped and passing the tests above under `-race`):
|
||
|
||
```go
|
||
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 (32 KiB)
|
||
reported bool // a Progress has been taken since the job started
|
||
finished time.Time // orders Pending
|
||
}
|
||
|
||
// Runner: mu guards jobs and queue; wake (cap 1) wakes the worker; changed
|
||
// (cap 1) tells the connection there is something to send. signal() does a
|
||
// non-blocking send so neither side ever blocks on the other.
|
||
type Runner struct {
|
||
exec Executor
|
||
timeout time.Duration
|
||
now func() time.Time
|
||
mu sync.Mutex
|
||
jobs map[string]*jobState
|
||
queue []string
|
||
wake chan struct{}
|
||
changed chan struct{}
|
||
}
|
||
|
||
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.Write` appends to both buffers under `mu`, then `signal(r.changed)`. `tailBuffer` keeps the last `max` bytes and a `truncated` flag. `TakeProgress` skips jobs that are not running, and running jobs that were already reported and have no new output. `Run` returns on ctx end; queued jobs are dropped (the master marks them lost on the next hello).
|
||
|
||
- [ ] **Step 4:** PASS with `-race`. Commit with W2-1.
|
||
|
||
### Task W2-4: Compose executor and override file
|
||
|
||
**Files:** Create `internal/agent/updater/override.go`, `override_test.go`, `resolve.go`, `resolve_test.go`; rewrite `compose_executor.go` (→ `compose.go` via `git mv`), add `compose_test.go`.
|
||
|
||
**Contracts:**
|
||
|
||
```go
|
||
// Resolver turns "repo:tag" into "repo:tag@digest" via the daemon.
|
||
type DistributionClient interface {
|
||
DistributionInspect(ctx context.Context, imageRef, encodedRegistryAuth string) (registry.DistributionInspect, error)
|
||
}
|
||
func NewResolver(cli DistributionClient, auth *RegistryAuth) *Resolver
|
||
// Resolve fails for a ref without a tag or with a digest already.
|
||
func (r *Resolver) Resolve(ctx context.Context, ref string) (string, error)
|
||
|
||
// CommandRunner runs "docker <args>" in dir with stdout and stderr to out.
|
||
type CommandRunner func(ctx context.Context, dir string, out io.Writer, args ...string) error
|
||
// ExecDocker is the real CommandRunner (exec.CommandContext, WaitDelay 10s).
|
||
func ExecDocker(ctx context.Context, dir string, out io.Writer, args ...string) error
|
||
|
||
// SelfInspector is the subset of the Docker SDK MountRoots needs.
|
||
type SelfInspector interface {
|
||
ContainerInspect(ctx context.Context, containerID string) (container.InspectResponse, error)
|
||
}
|
||
// MountRoots returns the agent container's read-write bind mounts whose
|
||
// host path equals the path inside the container.
|
||
func MountRoots(ctx context.Context, cli SelfInspector, selfID string) ([]string, error)
|
||
|
||
type ComposeDocker interface {
|
||
SelfInspector
|
||
ContainerList(ctx context.Context, opts container.ListOptions) ([]container.Summary, error)
|
||
}
|
||
func NewComposeExecutor(cli ComposeDocker, resolve *Resolver, selfID string, run CommandRunner) *ComposeExecutor
|
||
func (e *ComposeExecutor) Execute(ctx context.Context, job protocol.JobStart, out io.Writer) (string, error)
|
||
```
|
||
|
||
`ComposeExecutor.Execute` steps, in this order (each failure stops the job before the next step):
|
||
1. List containers (`All: true`) filtered by labels `com.docker.compose.project=<p>` and `com.docker.compose.service=<s>`; none → error `no container for compose service p/s on this machine`. Take `working_dir` and `config_files` labels from the first.
|
||
2. `MountRoots(ctx, cli, selfID)` (inspect error → job fails with it), then `composeFiles(roots, configFiles)` → `-f` list ending with the override, and the stack dir (code below).
|
||
3. `Resolve(job.Ref)` → `pinned`. Write `resolved <pinned>\n` to `out`.
|
||
4. `WriteOverride(dir, job.ComposeService, pinned)`.
|
||
5. `run(ctx, workingDir, out, "compose", "-p", p, "-f", f1, …, "-f", override, "pull", s)`, then the same with `"up", "-d", s`. Return `pinned`.
|
||
|
||
Override writer and path check (prototyped; tests below pass):
|
||
|
||
```go
|
||
// 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
|
||
}
|
||
|
||
// 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)
|
||
}
|
||
```
|
||
|
||
- [ ] **Step 1: Failing tests**
|
||
- `TestWriteOverride` (table, each case on its own `t.TempDir()`): new file → starts with header, contains `services.web.image`; existing ours with `db` entry → `db` kept, `web` set; existing ours with `web` → replaced; `compose.override.yaml` without header → refused, file unchanged; each of `compose.override.yml`, `docker-compose.override.yml`, `docker-compose.override.yaml` present → refused, ours not created; after success the directory holds no temp files.
|
||
- `TestMountRoots` (table, fake inspector): rw bind `/srv/stacks:/srv/stacks` → allowed; ro same-path bind → excluded; rw bind `/home/x/stacks:/stacks` (different path) → excluded; named volume → excluded; inspect error → error; `selfID ""` → the no-roots error.
|
||
- `TestComposeFiles` (table, roots are `t.TempDir()` paths): label `<root>/a/compose.yaml` → `[…/compose.yaml, …/compose.override.yaml]`; label already containing the override → listed once, last; nested path `<root>/apps/x/y/compose.yaml` → allowed; second root also allowed; empty label → error; no roots → error mentioning "no stack directory is mounted"; a file outside every root → exactly `stack directory <dir> is not mounted into the agent at the same path (read-write)`; `<root>/evil` symlinked to a dir outside → refused with the same message; `/foo-evil` vs root `/foo` → refused (lexical guard still applies); a root that is a file (socket stand-in) does not allow paths under it.
|
||
- `TestResolver` (fake `DistributionClient` recording args): `"gitea.shcizo.se/shcizo/slusk:1.4"` → called with that ref and the auth string for `gitea.shcizo.se`; returns `"gitea.shcizo.se/shcizo/slusk:1.4@sha256:d1"`; `"repo@sha256:x"` and `"repo"` → error without calling the daemon; daemon error is wrapped.
|
||
- `TestComposeExecutor` (fake Docker client returning containers and the agent's own mounts, fake resolver client, recording `CommandRunner`, temp dir as the mounted root): happy path runs exactly `compose -p slusk -f <root>/slusk/compose.yaml -f <root>/slusk/compose.override.yaml pull web` then `… up -d web` in `working_dir`, override written before the first command, returns pinned ref; no matching container → error, no commands; resolve fails → no override written, no commands; pull fails → `up` not run, error contains the command output; override refused → no commands; agent container mounts the root read-only → `not mounted` error, no override written, no commands.
|
||
- [ ] **Step 2:** FAIL. **Step 3:** Implement. **Step 4:** `go test -race ./internal/agent/updater` PASS.
|
||
|
||
### Task W2-5: Swarm executor, registry auth, dispatch
|
||
|
||
**Files:** `git mv internal/agent/updater/swarm_executor.go swarm.go` (+ test), create `registryauth.go`, `registryauth_test.go`, `dispatch.go`, `dispatch_test.go`.
|
||
|
||
**Contracts:**
|
||
|
||
```go
|
||
// RegistryAuth holds inline credentials from a docker config.json. A nil
|
||
// *RegistryAuth means anonymous access.
|
||
func LoadRegistryAuth(dockerConfigDir string, log *slog.Logger) (*RegistryAuth, error) // "" → nil, nil
|
||
// For returns the X-Registry-Auth value for repo's registry, or "".
|
||
func (a *RegistryAuth) For(repo string) string
|
||
|
||
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)
|
||
}
|
||
func NewSwarmExecutor(cli SwarmClient, resolve *Resolver, auth *RegistryAuth) *SwarmExecutor
|
||
|
||
// Dispatch picks the executor by job.Kind; KindContainer and unknown kinds fail.
|
||
func NewDispatch(compose, swarm Executor) Executor
|
||
```
|
||
|
||
Registry host mapping: `reference.ParseNormalizedNamed(repo)` → `reference.Domain(named)`; `docker.io` matches config keys `https://index.docker.io/v1/`, `index.docker.io`, `docker.io`, `registry-1.docker.io`; any other host matches its key with or without `https://`/`http://` and trailing path. Entries with `auth` (base64 `user:pass`) or `identitytoken` become `registry.EncodeAuthConfig(registry.AuthConfig{Username, Password, IdentityToken, ServerAddress})`. If the file has `credsStore` or `credHelpers`, log `registry_cred_helpers_unsupported` once.
|
||
|
||
`SwarmExecutor.Execute` steps:
|
||
1. `Info`: not `Swarm.ControlAvailable` → `refused: this agent's daemon is not a swarm manager`; `Swarm.Cluster == nil || Cluster.ID != job.SwarmClusterID` → `refused: job is for swarm cluster X, this daemon is in Y`. No inspect, no update.
|
||
2. `ServiceInspectWithRaw(job.SwarmService)`; check `svc.Spec.Name == job.SwarmService`.
|
||
3. `Resolve(job.Ref)` → pinned; write `resolved <pinned>\n`.
|
||
4. `spec := svc.Spec; spec.TaskTemplate.ContainerSpec.Image = pinned`; `ServiceUpdate(svc.ID, svc.Version, spec, opts)` with `QueryRegistry: true` and `EncodedRegistryAuth: auth.For(repo)` if non-empty, else `RegistryAuthFrom: swarm.RegistryAuthFromSpec`. Write any `resp.Warnings` to `out`. Return pinned.
|
||
|
||
- [ ] **Step 1: Failing tests**
|
||
- `TestSwarmExecutor` (table, fake client): happy path updates by ID `"svc-id"` with inspected version and image `repo:tag@sha256:…`; auth known → `EncodedRegistryAuth` set; unknown → `RegistryAuthFrom "spec"`; cluster mismatch → refused, zero `ServiceInspectWithRaw`/`ServiceUpdate` calls; worker node → refused; inspect returns a different name → error, no update; nil `ContainerSpec` → error.
|
||
- `TestRegistryAuth`: hub keys variants; `gitea.shcizo.se` with `https://gitea.shcizo.se` key; unknown host → `""`; empty dir → nil and `For` on nil returns `""`; malformed JSON → error.
|
||
- `TestDispatch`: routes by kind; container → error.
|
||
- [ ] **Step 2–4:** FAIL → implement → `go test -race ./internal/agent/updater` PASS. Commit W2-1..W2-5 together (see W2-1 message).
|
||
|
||
### Task W2-6: Connection — writer goroutine, job messages, resend
|
||
|
||
**Files:** `internal/agent/conn/conn.go`, `conn_test.go`
|
||
|
||
**Changes:**
|
||
- `New(cfg, src, jobs, log)` (C8). Hello: `ProtocolVersion = protocol.Version`, `KnownJobs = jobs.Known()`.
|
||
- One writer goroutine per session reading from `out chan protocol.Envelope` (cap 64); every send (hello, snapshot, upserts, removes, progress, results) goes through it. Each write uses a 10s timeout; a write error cancels the session.
|
||
- Reader goroutine: decodes envelopes. `job.start` with empty `ID` or undecodable payload → close session. Valid → `jobs.Submit(id, start)`. `job.ack` → `jobs.Ack(id)` and forget it from the session's `sent` set. Unknown type → `log.Warn("unknown_message_type", "type", …)`, continue.
|
||
- After the snapshot: `flush()` once, then the main loop also selects on `jobs.Changed()`. `flush()` sends `job.progress` for each `TakeProgress()` entry, then `job.result` for each `Pending()` entry not yet in this session's `sent` set. Throttle: at most one `flush()` per `ProgressInterval`; a signal arriving earlier arms a timer for the remainder instead of being dropped.
|
||
|
||
- [ ] **Step 1: Failing tests** (extend `fakeMaster` so the test can also *send* envelopes; fake `Jobs` records calls and returns scripted `Known`/`Pending`/`TakeProgress`):
|
||
- `TestAgent_HelloCarriesVersionAndKnownJobs`.
|
||
- `TestAgent_JobStartSubmitsAndAckAcks`.
|
||
- `TestAgent_SendsPendingResultsAfterSnapshot`: order hello → snapshot → result(s).
|
||
- `TestAgent_ResendsUnackedResultAfterReconnect`: result sent, master drops connection without ack, result sent again on the next session; after ack it is not resent on a third session.
|
||
- `TestAgent_ProgressIsThrottled`: 20 changed signals in 50ms with `ProgressInterval 100ms` → at most 2 progress batches in 150ms, and the last chunk is delivered.
|
||
- `TestAgent_IgnoresUnknownType`: master sends `{"type":"future.thing"}` → connection stays, a later `job.start` still works.
|
||
- `TestAgent_ClosesOnMalformedJobStart`: `job.start` without ID → agent reconnects (new connection observed).
|
||
- Run with `-race`; one test sends inventory changes and job progress concurrently.
|
||
- [ ] **Step 2–4:** FAIL → implement → `go test -race ./internal/agent/...` PASS.
|
||
- [ ] **Step 5: Commit** — `feat(agent): run jobs from the master and resend results until acked (#7)`
|
||
|
||
### Task W2-7: Agent binary, image, examples
|
||
|
||
**Files:** `cmd/agent/main.go`, `Dockerfile`, delete `docker-compose.example.yml`, `docker-compose.local.yml`, `env.sample`; create `deploy/agent.compose.yml`.
|
||
|
||
- [ ] **Step 1:** `cmd/agent` wiring:
|
||
|
||
```go
|
||
auth, err := updater.LoadRegistryAuth(cfg.DockerConfig, logger)
|
||
resolver := updater.NewResolver(dockerCli, auth)
|
||
exec := updater.NewDispatch(
|
||
updater.NewComposeExecutor(dockerCli, resolver, cfg.SelfID, updater.ExecDocker),
|
||
updater.NewSwarmExecutor(dockerCli, resolver, auth),
|
||
)
|
||
runner := updater.NewRunner(exec, cfg.UpdateTimeout)
|
||
go runner.Run(ctx)
|
||
agent := conn.New(conn.Config{...}, collector, runner, logger)
|
||
```
|
||
|
||
Log `agent_starting` with `update_timeout`, `registry_auth` (bool) — never the token. At startup call `updater.MountRoots(ctx, dockerCli, cfg.SelfID)` once, only to log: on error or zero roots log `no_stack_mounts` at warn ("Compose updates will fail; mount each stacks directory at the same path, read-write"), else log `stack_mounts` with the roots. Jobs read the mounts again themselves.
|
||
- [ ] **Step 2: Dockerfile:** build only `master:updater-master agent:updater-agent`; `EXPOSE 8080 8090`; `ENTRYPOINT ["/usr/local/bin/updater-agent"]`. Why the agent: it is the image's many-copies deployment (one per machine, pasted from the UI), the master stack already sets its own entrypoint, and an old container that accidentally picks up the new image fails fast on missing `MASTER_URL` instead of serving anything.
|
||
- [ ] **Step 3:** `deploy/agent.compose.yml` = the UI snippet (W4-4) with placeholders and comments.
|
||
- [ ] **Step 4:** `go build ./cmd/agent && docker build -t package-updater:dev .` then `docker run --rm package-updater:dev` exits with `fatal: MASTER_URL is required`. Remove the built `./agent` binary.
|
||
- [ ] **Step 5: Commit** — `build: agent is the image's default entrypoint; drop the old server examples (#7)`
|
||
|
||
---
|
||
|
||
## W3 — Master
|
||
|
||
Packages W3 tests while W2–W4 run in parallel: `go vet ./internal/master/store/ ./internal/master/hub/ ./internal/master/jobs/ ./cmd/master/ && go test -race ./internal/master/store/ ./internal/master/hub/ ./internal/master/jobs/` (plus the `integration` tag for the store when `DATABASE_URL` is set).
|
||
|
||
### Task W3-1: Store — jobs, config files, desired ref
|
||
|
||
**Files:** Create `internal/master/store/jobs.go`, `jobs_integration_test.go`; modify `services.go`, `agents.go`, `services_integration_test.go`, `agents_integration_test.go`.
|
||
|
||
- [ ] **Step 1: Failing integration tests** (`//go:build integration`):
|
||
- `TestCreateJob_OneActivePerService`: two goroutines `CreateJob` for the same service concurrently → exactly one succeeds; the other gets `model.ErrJobActive` and the same job ID. After `FailJob`, a new `CreateJob` succeeds.
|
||
- `TestFailLostJobs`: agent A has jobs 1 (queued), 2 (running), 3 (succeeded); agent B has 4 (running). `FailLostJobs(A, [2])` → returns `[1]`; 1 failed with the "lost" message; 2, 3, 4 unchanged.
|
||
- `TestJobProgress_OwnerOnly`: progress from A moves queued → running and sets `started_at`; progress for the same job from B returns false and changes nothing; progress after finish returns false.
|
||
- `TestJobProgress_KeepsTail`: appending beyond `MaxJobOutput` keeps the last `MaxJobOutput` characters and sets `output_truncated`.
|
||
- `TestFinishJob_OnlyOnceAndOnlyByOwner`: B → false; A succeeded with `Ref "r:1@sha256:d"` → true, job `to_ref` and service `desired_ref` both `r:1@sha256:d`, output replaced; second call → false, nothing changes; a failed finish leaves `desired_ref` untouched.
|
||
- `TestGetJobAndListJobs`: `ServiceName` filled from the service row; newest first; `ServiceID` filter; `Limit`; list has empty `Output`.
|
||
- `TestGetService_IncludesRemoved`; snapshot round-trips `ConfigFiles`.
|
||
- `TestRevokeAgent_FailsActiveJobs`: queued/running jobs of the agent → failed "agent revoked"; other agents' jobs untouched.
|
||
- [ ] **Step 2:** `DATABASE_URL=… go test -tags integration ./internal/master/store/` — FAIL.
|
||
- [ ] **Step 3: Implement.** The two statements with subtle semantics:
|
||
|
||
```sql
|
||
-- JobProgress ($1 job, $2 agent, $3 chunk, $4 at, $5 = protocol.MaxJobOutput)
|
||
UPDATE update_jobs SET
|
||
status = 'running',
|
||
started_at = COALESCE(started_at, $4),
|
||
output_truncated = output_truncated OR length(output) + length($3) > $5,
|
||
output = right(output || $3, $5)
|
||
WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running')
|
||
|
||
-- FinishJob, inside one transaction
|
||
UPDATE update_jobs SET
|
||
status = $3, error = $4, output = $5, output_truncated = $6,
|
||
to_ref = CASE WHEN $7 <> '' THEN $7 ELSE to_ref END,
|
||
started_at = COALESCE(started_at, $8), finished_at = $9
|
||
WHERE id = $1 AND agent_id = $2 AND status IN ('queued', 'running')
|
||
RETURNING service_id
|
||
-- then, only if a row came back and $3 = 'succeeded':
|
||
UPDATE services SET desired_ref = $7 WHERE id = $service_id
|
||
```
|
||
|
||
`CreateJob`: plain `INSERT … RETURNING`; on a unique violation of `update_jobs_one_active` (`pgconn.PgError` code `23505`, constraint name) select the active job for the service and return it with `model.ErrJobActive`.
|
||
- [ ] **Step 4:** PASS. `go vet ./... && go test ./...` (non-integration) still PASS.
|
||
- [ ] **Step 5: Commit** — `feat(master): store update jobs (#7)`
|
||
|
||
### Task W3-2: Hub — per-connection writer, routing, reconcile, job messages
|
||
|
||
**Files:** `internal/master/hub/hub.go` (split into `hub.go`, `route.go`, `jobs.go` if it helps), `hub_test.go`, new `route_test.go`.
|
||
|
||
**Changes:**
|
||
- `conns map[int64]*agentConn` where `agentConn{ws, id uint64 (from an atomic counter), out chan protocol.Envelope (cap 64), ctx, cancel, hello *protocol.Hello}`. A writer goroutine per connection does every write (10s timeout per write; on error `CloseNow` + cancel). `Online` and `Disconnect` keep their meaning.
|
||
- `Store` interface gains `FailLostJobs`, `JobProgress`, `FinishJob` (C5 signatures).
|
||
- Hello handling, in this order: `RecordHello` → parse `KnownJobs` (skip non-integers with a warning) → `FailLostJobs` → publish `Jobs` and each lost `JobTopic` → under `h.mu`, set `conn.hello` (from now on the connection is routable) → publish `Agents`. A connection is never routable before its reconcile ran.
|
||
- `Route(svc)`: copy the routable connections under `h.mu`, then:
|
||
|
||
```go
|
||
// route picks the connection a job for svc goes to. Compose services and
|
||
// containers can only run on the agent that reported them. Swarm services
|
||
// run on the reporting agent if it qualifies, else on any other connected
|
||
// manager of the same cluster (lowest agent ID first). Too-old wins over
|
||
// offline only when a connection would otherwise have qualified.
|
||
func route(svc model.Service, conns []candidate) (model.Route, error) {
|
||
sort.Slice(conns, func(a, b int) bool { return conns[a].agentID < conns[b].agentID })
|
||
tooOld := false
|
||
try := func(c candidate) bool {
|
||
if svc.Kind == protocol.KindSwarm {
|
||
if !c.hello.SwarmManager || c.hello.SwarmClusterID == "" || c.hello.SwarmClusterID != svc.SwarmClusterID {
|
||
return false
|
||
}
|
||
}
|
||
if c.hello.ProtocolVersion < protocol.MinJobVersion {
|
||
tooOld = true
|
||
return false
|
||
}
|
||
return true
|
||
}
|
||
if svc.AgentID != nil {
|
||
for _, c := range conns {
|
||
if c.agentID == *svc.AgentID && try(c) {
|
||
return model.Route{AgentID: c.agentID, ConnID: c.connID}, nil
|
||
}
|
||
}
|
||
}
|
||
if svc.Kind == protocol.KindSwarm {
|
||
for _, c := range conns {
|
||
if try(c) {
|
||
return model.Route{AgentID: c.agentID, ConnID: c.connID}, nil
|
||
}
|
||
}
|
||
}
|
||
if tooOld {
|
||
return model.Route{}, model.ErrAgentTooOld
|
||
}
|
||
return model.Route{}, model.ErrAgentOffline
|
||
}
|
||
```
|
||
|
||
- `StartJob(r, jobID, start)`: under `h.mu` look up `conns[r.AgentID]`; missing, `id != r.ConnID` or `hello == nil` → `model.ErrAgentOffline`. Otherwise enqueue `EncodeJob(TypeJobStart, strconv.FormatInt(jobID,10), start)` (blocking on `out` or the connection's ctx; ctx done → `ErrAgentOffline`). An enqueued job whose write later fails stays queued and is settled by the next hello's reconcile.
|
||
- `job.progress`: non-integer ID → warn, ignore. `JobProgress(agentID, id, payload.Output, now)`; applied → publish `Jobs`, `JobTopic(id)`.
|
||
- `job.result`: non-integer ID → warn, ack, return. Status other than `succeeded`/`failed` → store as failed with error `agent sent invalid status "<x>"`. `FinishJob(agentID, id, …)`: store error → return it (closes the connection; agent resends after reconnect). Not applied → warn `job_result_ignored`. Always enqueue `job.ack` afterwards. Applied → publish `Jobs`, `JobTopic(id)`, and `Services` if succeeded.
|
||
- Unknown type (after hello) → warn `unknown_message_type`, continue. Before hello any non-hello still closes (phase 1 rule).
|
||
|
||
- [ ] **Step 1: Failing tests** (extend `fakeStore` with the three methods, recording calls and returning scripted results):
|
||
- `TestRoute` (table on the pure `route` function): compose on own agent v2 → ok; own agent v0 → `ErrAgentTooOld`; own agent missing → `ErrAgentOffline`; another agent online for a compose service → still offline; swarm: reporting agent qualifies → it; reporting agent offline, other manager same cluster v2 → the other; other manager different cluster → offline; only a v0 manager of the cluster → too old; worker of the cluster → offline; manager with empty cluster ID → offline.
|
||
- `TestHub_ReconcilesBeforeRoutable`: fake store blocks `FailLostJobs` until released; while blocked, `Route` for the agent's service returns `ErrAgentOffline`; after release it routes. `FailLostJobs` got `known = [7, 9]` from `KnownJobs ["7","9","x"]`.
|
||
- `TestHub_StartJobSendsEnvelope`: client receives `job.start` with `ID "42"` and the payload.
|
||
- `TestHub_StartJobOnReplacedConnectionFails`: route, then the agent reconnects (new connection), `StartJob` with the old route → `ErrAgentOffline`, nothing written to the new connection.
|
||
- `TestHub_ProgressAndResult`: progress → `JobProgress(agent, 42, "out", …)` and `job-42` published; result → `FinishJob`, ack received with `ID "42"`; not-applied result → still acked; result with ID `"abc"` → acked, no store call; result with status `"weird"` → stored as failed.
|
||
- `TestHub_ResultStoreErrorClosesConnection`.
|
||
- `TestHub_IgnoresUnknownType`: `{"type":"future.thing"}` after hello → connection stays open, a following upsert is stored.
|
||
- `TestHub_LargeResultArrives`: a result whose `Output` is `protocol.MaxJobOutput` bytes of `"\x1b"` reaches `FinishJob` intact.
|
||
- `TestHub_ConcurrentWritesAreSerialised` (`-race`): 50 `StartJob` calls from goroutines while the client sends 50 results (acks written concurrently) — all 50 job.starts and 50 acks arrive, each a valid JSON envelope.
|
||
- Existing phase-1 hub tests keep passing.
|
||
- [ ] **Step 2–4:** FAIL → implement → `go test -race ./internal/master/hub` PASS.
|
||
- [ ] **Step 5: Commit** — `feat(master): route jobs to agents and reconcile them on reconnect (#7)`
|
||
|
||
### Task W3-3: Jobs service
|
||
|
||
**Files:** Create `internal/master/jobs/jobs.go`, `jobs_test.go`.
|
||
|
||
**Contract:** C6. Rules, in this order:
|
||
|
||
```go
|
||
// Updatable returns nil or a *model.NotUpdatableError.
|
||
func (s *Service) Updatable(svc model.Service) error {
|
||
switch {
|
||
case svc.RemovedAt != nil:
|
||
return &model.NotUpdatableError{Reason: model.ReasonRemoved}
|
||
case svc.Self || imageref.Parse(svc.ActualRef).Repo == s.selfRepo:
|
||
return &model.NotUpdatableError{Reason: model.ReasonSelf}
|
||
case svc.Kind == protocol.KindContainer:
|
||
return &model.NotUpdatableError{Reason: model.ReasonContainer}
|
||
case imageref.Parse(svc.ActualRef).Tag == "":
|
||
return &model.NotUpdatableError{Reason: model.ReasonNoTag}
|
||
}
|
||
return nil
|
||
}
|
||
```
|
||
|
||
`selfRepo = imageref.Parse(agentImage).Repo` (master and agent are the same image; `AGENT_IMAGE` defaults to `gitea.shcizo.se/shcizo/package-updater:latest`).
|
||
|
||
`Trigger`: `GetService` → `Updatable` → `route, routeErr := hub.Route(svc)` → `CreateJob(NewJob{ServiceID, AgentID: &route.AgentID or nil, FromRef: svc.ActualRef, ToRef: repo:tag, Trigger: manual, RequestedBy, CreatedAt: now})` (on `ErrJobActive` return it as is) → if `routeErr != nil`: `FailJob(job.ID, routeErr.Error())` → else `StartJob(route, job.ID, start)`; on error `FailJob(job.ID, model.ErrAgentOffline.Error())` → publish `Jobs` + `JobTopic` → return `GetJob(job.ID)` (so the caller sees the final status). `JobStart` for compose: `Kind, ComposeProject, ComposeService, Ref`; for swarm: `Kind, SwarmClusterID, SwarmService, Ref`.
|
||
|
||
- [ ] **Step 1: Failing tests**
|
||
- `TestUpdatable` (table): removed; `Self`; same repo as `AGENT_IMAGE` with a different tag (`…/package-updater:1.2.0@sha256:…`); container; `repo@sha256:…` (no tag); `postgres:17.10` → nil; `localhost:5000/foo:v1` → nil.
|
||
- `TestTrigger` (table, fake store + fake hub + fake publisher): happy compose → job queued, hub got exact `JobStart` with `Ref "gitea.shcizo.se/shcizo/slusk:1.4.2"` from `ActualRef "gitea.shcizo.se/shcizo/slusk:1.4.2@sha256:d1"`, `FromRef` = full actual ref, `RequestedBy` passed, `Jobs` published; swarm → `SwarmClusterID`/`SwarmService` set; not updatable → `*NotUpdatableError`, no job created; active job → that job + `ErrJobActive`, no hub call; route offline → job created and failed "agent offline", no `StartJob`; route too old → failed "agent too old for updates"; `StartJob` fails → failed "agent offline"; unknown service → `ErrNotFound`.
|
||
- [ ] **Step 2–4:** FAIL → implement → PASS.
|
||
- [ ] **Step 5: Commit** — `feat(master): start manual update jobs with self-protection (#7)`
|
||
|
||
### Task W3-4: Master wiring
|
||
|
||
**Files:** `cmd/master/main.go`.
|
||
|
||
- [ ] **Step 1:** Wire per C7, replacing W4-0's temporary `nil` with `jobService`. No new env vars (`AGENT_IMAGE` exists).
|
||
- [ ] **Step 2:** `go build ./cmd/master` (W4-0 is committed before this task starts, so `web.New` already has the C7 signature). Remove the built binary.
|
||
- [ ] **Step 3: Commit** — `feat(master): wire jobs into the master (#7)`
|
||
|
||
---
|
||
|
||
## W4 — Web
|
||
|
||
W4 works against C5/C6/C7 with fakes; it does not need W3's code to compile its own package. Integration point: `cmd/master` (W3-4) passes `*jobs.Service` and `*store.Store`.
|
||
|
||
Packages W4 tests while W2–W4 run in parallel: `go vet ./internal/master/web/ ./internal/master/webauth/ && go test ./internal/master/web/ ./internal/master/webauth/`.
|
||
|
||
### Task W4-0: Constructor signature first
|
||
|
||
**Files:** `internal/master/web/server.go`, `server_test.go`, `cmd/master/main.go` (this one call only — the single exception to W3 owning `cmd/master`; tell the W3 owner when it is committed).
|
||
|
||
- [ ] **Step 1:** Add the `Jobs` interface and the three `Store` methods from C7, change `New` to `New(store, hub, jobs, broker, cfg, log)` and store `jobs` on `Server`. No routes use it yet. Extend `fakeStore` with the three methods and add a `fakeJobs` in `server_test.go`.
|
||
- [ ] **Step 2:** In `cmd/master/main.go`, change only the `web.New` call: pass `nil` as `jobs`, with the comment `// wired in W3-4`. This is the one temporary value in the plan, and it is there only so the shared module builds at every commit: the real `*jobs.Service` cannot be built until W3-1..W3-3 exist (it needs `*store.Store` and `*hub.Hub` to satisfy their interfaces). No handler calls `s.jobs` before W4-1, and W3-4 replaces the `nil` before W5 starts; the final verification greps that the `// wired in W3-4` comment is gone.
|
||
- [ ] **Step 3:** `go build ./cmd/master ./internal/master/web/ && go test ./internal/master/web/` PASS.
|
||
- [ ] **Step 4: Commit** — `refactor(web): web.New takes the jobs service (#7)`
|
||
|
||
### Task W4-1: Service page and Update action
|
||
|
||
**Files:** `internal/master/web/server.go`, `views.go`, new `templates/service.html`, `templates/overview.html`, `templates/layout.html`, `static/app.css`, `server_test.go`; `internal/master/webauth/auth.go` + `auth_test.go`.
|
||
|
||
**Routes:**
|
||
|
||
| Route | Does |
|
||
|---|---|
|
||
| `GET /services/{id}` | details (name, kind, agent, image `ActualRef`, digest, status, desired ref, config files split on `,`), Update button or the disabled reason from `jobs.Updatable`, job history (`ListJobs{ServiceID, Limit: 20}`) |
|
||
| `GET /fragments/services/{id}/jobs` | the history table only; `hx-trigger="sse:jobs"` |
|
||
| `POST /services/{id}/update` | CSRF hidden field (as revoke); `jobs.Trigger(ctx, id, sess.Subject)`; success or `ErrJobActive` → `303` to `/jobs/{job.ID}`; `*NotUpdatableError` → `409` service page with the reason; `ErrNotFound` → `404`; other → `500` |
|
||
|
||
`webauth`: add `// WithSession returns ctx carrying sess, as Require does.` `func WithSession(ctx context.Context, sess model.Session) context.Context`, and make `Require` use it. Tests use it in their `protect` wrapper to inject a session with `Subject "alice"`.
|
||
|
||
Overview: the service name links to `/services/{id}`; nothing else on the overview changes. Nav: `Services · Jobs · Agents`.
|
||
|
||
- [ ] **Step 1: Failing tests** (fake store gains `GetService`, `GetJob`, `ListJobs`; fake `Jobs` returns scripted `Updatable` and `Trigger` results and records `requestedBy`):
|
||
- `TestServicePage_ShowsUpdateButton` (updatable) and `TestServicePage_DisabledReason` (table over the four reasons: page shows the reason text, no enabled Update button).
|
||
- `TestServicePage_NotFound` → 404; bad ID → 400.
|
||
- `TestUpdate_RedirectsToJob` (`requestedBy == "alice"`), `TestUpdate_ActiveJobRedirectsToExisting`, `TestUpdate_NotUpdatableShows409`.
|
||
- `TestServiceJobsFragment_IsNotAFullPage`.
|
||
- `TestOverview_LinksToServicePage`.
|
||
- [ ] **Step 2–4:** FAIL → implement → PASS.
|
||
- [ ] **Step 5: Commit** — `feat(web): service page with manual update (#7)`
|
||
|
||
### Task W4-2: Jobs list and live job log
|
||
|
||
**Files:** `server.go`, new `templates/jobs.html`, `templates/job.html`, `server_test.go`.
|
||
|
||
| Route | Does |
|
||
|---|---|
|
||
| `GET /jobs` | `ListJobs{Limit: 100}`: id, service (link), status, from → to, requested by, created, duration |
|
||
| `GET /fragments/jobs` | table only; `hx-trigger="sse:jobs"` |
|
||
| `GET /jobs/{id}` | job details, error, `<pre>` with `Output` (plus "earlier output truncated" when `OutputTruncated`) |
|
||
| `GET /fragments/jobs/{id}` | details + log; `hx-trigger="sse:job-{{.Job.ID}}"`. Always re-reads the job from the store: the DB is the source of truth, so a dropped signal or SSE reconnect just means the next fetch catches up. |
|
||
|
||
- [ ] **Step 1: Failing tests:** list renders statuses and links; job page shows output and error; truncated note; fragment is not a full page and contains `sse:job-42`; unknown job → 404.
|
||
- [ ] **Step 2–4:** FAIL → implement → PASS.
|
||
- [ ] **Step 5: Commit** — `feat(web): jobs list and live job log (#7)`
|
||
|
||
### Task W4-3: Revoke publishes jobs
|
||
|
||
**Files:** `server.go`, `server_test.go`.
|
||
|
||
- [ ] `revokeAgent` also publishes `events.Jobs` (the store failed the agent's active jobs). Test: `TestRevokeAgent_RevokesAndDisconnects` asserts both topics. **Commit** with W4-4.
|
||
|
||
### Task W4-4: Agent snippet
|
||
|
||
**Files:** `server.go` (`agentSnippet`), `server_test.go`.
|
||
|
||
```yaml
|
||
services:
|
||
updater-agent:
|
||
image: <AgentImage>
|
||
entrypoint: ["/usr/local/bin/updater-agent"]
|
||
restart: unless-stopped
|
||
environment:
|
||
MASTER_URL: <AgentURL>
|
||
MASTER_KEY: <MasterKey>
|
||
AGENT_TOKEN: <token>
|
||
UPDATE_TIMEOUT: 5m
|
||
# Optional registry credentials (a config.json written by docker login):
|
||
# DOCKER_CONFIG: /docker-config
|
||
volumes:
|
||
- /var/run/docker.sock:/var/run/docker.sock
|
||
# - /path/to/stacks:/path/to/stacks # mount each stacks directory at the same path, read-write
|
||
# - /root/.docker/config.json:/docker-config/config.json:ro
|
||
```
|
||
|
||
- [ ] Test: snippet does **not** contain `STACKS_ROOT`; contains the commented same-path mount example line exactly as above, `UPDATE_TIMEOUT`, the commented `DOCKER_CONFIG`. **Commit** — `feat(web): agent snippet mounts stacks read-write for updates (#7)`
|
||
|
||
---
|
||
|
||
## W5 — Integration (after W2–W4 are on the branch)
|
||
|
||
### Task W5-1: End-to-end jobs
|
||
|
||
**Files:** `internal/e2e/agent_master_test.go` (split into `jobs_test.go` if long).
|
||
|
||
- [ ] **Step 1:** Extend `memStore` with in-memory `GetService`, `CreateJob` (one active per service), `FailJob`, `GetJob`, `FailLostJobs`, `JobProgress`, `FinishJob` (same ownership and "only active" rules as C5). `startMaster` returns a `restart()` that builds a new `hub.New` over the **same** `memStore`, swaps it into the mux through an `atomic.Pointer[hub.Hub]`, and calls `srv.CloseClientConnections()` — the hub holds no job state, so this is a faithful master restart. Agents use a real `updater.Runner` with a fake `Executor` that writes output and blocks until the test releases it.
|
||
- [ ] **Step 2: Tests**
|
||
- `TestManualJobFlowsToAgentAndBack`: compose service in the store, `jobs.Service.Trigger` → job succeeded, `to_ref`/`desired_ref = "<ref>@sha256:e2e"`, output contains the executor's lines, runner `Known()` empty (acked).
|
||
- `TestJobFinishedWhileDisconnectedIsRecordedOnce`: start job, `CloseClientConnections` while the executor blocks, release, agent reconnects → `FinishJob` applied exactly once (count calls that returned true), runner empty afterwards.
|
||
- `TestMasterRestartKeepsJobTheAgentKnows`: start job, `restart()` while it runs, release → job succeeded, never failed "lost".
|
||
- `TestAgentRestartMarksJobLost`: start job, cancel the agent's `Run` and drop its runner, start a new agent (new runner) with the same token → job failed with the "lost" message.
|
||
- `TestPhaseOneAgentGetsNoJobs`: a raw websocket client sends a phase-1 hello (no `protocol_version`) + snapshot; `Trigger` → job failed "agent too old for updates"; the client receives no `job.start` within 200ms.
|
||
- [ ] **Step 3:** `go test -race ./internal/e2e/` PASS.
|
||
- [ ] **Step 4: Commit** — `test: jobs end to end over pinned TLS (#7)`
|
||
|
||
### Task W5-2: Documentation
|
||
|
||
**Files:** `CLAUDE.md`, `README.md`, `deploy/master.stack.yml`.
|
||
|
||
- [ ] `CLAUDE.md`:
|
||
- Intro: master/agent fleet updater (no longer "webhook-driven, single binary").
|
||
- Commands: drop `go build ./cmd/server` and the local smoke test (`docker-compose.local.yml`, `env.sample` are gone); keep `go build ./cmd/master ./cmd/agent`, integration tests, docker build.
|
||
- Architecture: remove `cmd/server`, `internal/api`, `internal/config`, `internal/discovery`, `internal/updater`, `internal/selfupdate`, `internal/metrics`; add `internal/agent/updater` (runner + Compose/Swarm executors + override file + mount-root check + registry auth) and `internal/master/jobs`.
|
||
- Design intents: remove "Defense in depth: token AND opt-in label AND STACKS_ROOT", "Auth on /update only", "metrics may be nil", "Stateless", "MODE"; replace with: single FIFO worker **per agent** (`internal/agent/updater/runner.go`); one active job per service (unique index); the agent writes only `compose.override.yaml` with the header and refuses when any other override name exists; every config file and the override path must be inside a read-write, same-path bind mount of the agent's own container (read per job, checked after symlink resolution) — replaces `STACKS_ROOT`; self-protection by `AGENT_IMAGE` repo + `self` flag; unknown message types ignored, malformed messages close.
|
||
- Gotchas: remove the selfupdate, `MODE`, `STACKS_ROOT` prod-default, "stateless/label gate describe cmd/server only" entries; update the `IsInsideRoot` path; add: Compose loads only one override file; stacks started with explicit `-f` don't load the override on a manual `up`; override entries are never pruned; there is no `STACKS_ROOT`: stacks directories must be mounted rw at the host path, and `SELF_ID` must resolve to the agent's container (on Docker Desktop/OrbStack bind sources are translated, so Compose jobs fail there); `job.start` carries no paths; default entrypoint is the agent.
|
||
- [ ] `README.md`: replace the webhook-era sections (trigger flow, opt-in label, Swarm `MODE`, quick start with `UPDATER_API_KEY`) with: what master and agents do, deploying the master (`deploy/master.stack.yml`), adding agents (UI snippet / `deploy/agent.compose.yml`), the agent env table (C8), mounting stacks directories (`/path:/path`, read-write, one line per directory; no `STACKS_ROOT`), how Compose updates work and their limits (override file, explicit `-f`, operator overrides refused, no pruning), a "check before enabling" snippet listing every stack's config files so the operator can spot existing override files and explicit `-f` stacks:
|
||
|
||
```bash
|
||
docker ps --format '{{.Label "com.docker.compose.project.config_files"}}' | sort -u
|
||
```
|
||
|
||
Swarm updates, and "the webhook returns in phase 3; until then the Gitea action has nothing to call".
|
||
- [ ] `deploy/master.stack.yml`: header comment "phase 2: inventory and manual updates".
|
||
- [ ] **Commit** — `docs: master/agent phase 2 — jobs and manual update (#7)`
|
||
|
||
---
|
||
|
||
## Final verification
|
||
|
||
Run from the worktree root:
|
||
|
||
```bash
|
||
go mod tidy && git diff --exit-code go.mod go.sum # W2 may have left tidy to here; commit if it changed
|
||
! grep -n "wired in W3-4" cmd/master/main.go # W4-0's temporary nil is gone
|
||
go vet ./... && go test ./... && go test -race ./internal/agent/... ./internal/master/... ./internal/e2e/ \
|
||
&& go build ./cmd/master ./cmd/agent && docker build -t package-updater:dev .
|
||
rm -f master agent
|
||
DATABASE_URL=postgres://... go test -tags integration ./internal/master/store/ # when a Postgres is available
|
||
docker run --rm package-updater:dev # expect: fatal: MASTER_URL is required
|
||
```
|
||
|
||
Manual smoke test (Linux machine, a throwaway Compose stack in a directory mounted into the agent at the same path, read-write): deploy the master, add an agent, open the stack's service page, click *Update*, watch the log stream, confirm `compose.override.yaml` appeared with the header and `repo:tag@digest`, run `docker compose up -d` by hand in the stack dir and confirm the container keeps the pinned digest.
|