diff --git a/README.md b/README.md index 8129669..1f5ff8e 100644 --- a/README.md +++ b/README.md @@ -102,6 +102,7 @@ All JSON, all under `/api/v1`, all on the loopback listener. | `GET` | `/api/v1/services` | Every service with its running, merged, and latest-available digests, drift flag, and last receipt. | | `POST` | `/api/v1/services/{name}/deploy` | `{"digest": "sha256:…"}` → `{"journalId": N}`, `202`. | | `POST` | `/api/v1/services/{name}/rollback` | `{"digest": "sha256:…"}` (optional) → `{"journalId": N}`, `202`. | +| `POST` | `/api/v1/services/{name}/converge` | No body. Re-applies the merged manifest without moving the pin → `{"journalId": N}`, `202`. | | `GET` | `/api/v1/services/{name}/history?n=20` | Newest-first receipts. | | `GET` | `/api/v1/events` | Server-sent `state` events for every transition. | | `GET` | `/healthz` | `200`, no identity required. | @@ -122,14 +123,23 @@ address. snowdeploy status snowdeploy deploy [--digest sha256:…] snowdeploy rollback [--to sha256:…] +snowdeploy converge snowdeploy history [-n 20] snowdeploy validate [--volume-prefixes p1,p2] [--env-file-prefixes p1,p2] ``` `--server` and `--token-file` (or `SNOWDEPLOY_SERVER` and -`SNOWDEPLOY_TOKEN_FILE`) point it at a daemon. `deploy` and `rollback` follow -the event stream and exit non-zero unless the service ends healthy, so they -work in a script. +`SNOWDEPLOY_TOKEN_FILE`) point it at a daemon. `deploy`, `rollback`, and +`converge` follow the event stream and exit non-zero unless the service ends +healthy, so they work in a script. + +`converge` exists for the manifest change a deploy cannot carry: an env, +volume, or template edit that merged without moving the image digest. It +re-renders and re-applies the manifest as main has it — no pull request, since +the change already went through review — and, unlike a deploy, it has no +automatic rollback: the previously rendered unit is recorded nowhere, so a +failed probe leaves the run `failed` with instructions rather than silently +restoring anything. ## Security model diff --git a/cmd/snowdeploy/client.go b/cmd/snowdeploy/client.go index a4aff16..bef9f2c 100644 --- a/cmd/snowdeploy/client.go +++ b/cmd/snowdeploy/client.go @@ -200,7 +200,10 @@ func (c *client) startDeploy( JournalID int64 `json:"journalId"` } path := fmt.Sprintf("/api/v1/services/%s/%s", url.PathEscape(service), action) - body := map[string]string{"digest": digest} + var body any + if action != "converge" { + body = map[string]string{"digest": digest} + } if err := c.do(ctx, http.MethodPost, path, body, &out); err != nil { return 0, err } diff --git a/cmd/snowdeploy/client_test.go b/cmd/snowdeploy/client_test.go index 9a0d2fd..e6c8233 100644 --- a/cmd/snowdeploy/client_test.go +++ b/cmd/snowdeploy/client_test.go @@ -3,6 +3,7 @@ package main import ( "encoding/json" "fmt" + "io" "net/http" "net/http/httptest" "os" @@ -145,6 +146,8 @@ type fakeDaemon struct { deployPath string authSeen string rollbackHit bool + convergeHit bool + convergeRaw []byte journalID int64 states []string @@ -218,6 +221,17 @@ func (f *fakeDaemon) handler(t *testing.T) http.Handler { f.begun(8) }) + mux.HandleFunc("POST /api/v1/services/{name}/converge", func(w http.ResponseWriter, r *http.Request) { + raw, _ := io.ReadAll(r.Body) + f.mu.Lock() + f.convergeHit = true + f.convergeRaw = raw + f.deployPath = r.URL.Path + f.mu.Unlock() + writeJSON(t, w, map[string]int64{"journalId": 9}) + f.begun(9) + }) + mux.HandleFunc("GET /api/v1/events", func(w http.ResponseWriter, r *http.Request) { w.Header().Set("Content-Type", "text/event-stream") w.WriteHeader(http.StatusOK) @@ -373,6 +387,51 @@ func TestRollbackWithoutTarget(t *testing.T) { } } +func TestConvergePostsNoBodyAndFollowsToHealthy(t *testing.T) { + f, url := newFakeDaemon(t, "reconciling", "probing", "healthy") + + code, c := runCLI("converge", "web", "--server", url) + if code != 0 { + t.Fatalf("exit = %d, want 0\nstdout: %s\nstderr: %s", + code, c.out.String(), c.err.String()) + } + + f.mu.Lock() + defer f.mu.Unlock() + if !f.convergeHit { + t.Fatal("converge endpoint not called") + } + if f.deployPath != "/api/v1/services/web/converge" { + t.Errorf("posted to %q", f.deployPath) + } + if len(f.convergeRaw) != 0 { + t.Errorf("converge sent a body it has no use for: %q", f.convergeRaw) + } + for _, want := range []string{"reconciling", "probing", "healthy"} { + if !strings.Contains(c.out.String(), want) { + t.Errorf("output missing state %q:\n%s", want, c.out.String()) + } + } +} + +func TestConvergeExitsNonZeroOnFailure(t *testing.T) { + _, url := newFakeDaemon(t, "reconciling", "failed") + + code, c := runCLI("converge", "web", "--server", url) + if code == 0 { + t.Fatalf("a failed converge exited 0:\n%s", c.out.String()) + } +} + +func TestConvergeRefusesADigestFlag(t *testing.T) { + _, url := newFakeDaemon(t) + + code, _ := runCLI("converge", "web", "--digest", "sha256:aaa", "--server", url) + if code != 2 { + t.Fatalf("converge with --digest = %d, want the usage exit code 2", code) + } +} + func TestHistoryPrintsEntries(t *testing.T) { _, url := newFakeDaemon(t) code, c := runCLI("history", "web", "--server", url) diff --git a/cmd/snowdeploy/main.go b/cmd/snowdeploy/main.go index 2585817..645004d 100644 --- a/cmd/snowdeploy/main.go +++ b/cmd/snowdeploy/main.go @@ -62,6 +62,8 @@ func run(args []string, stdout, stderr io.Writer) int { return followedCommand(ctx, "deploy", rest, stdout, stderr) case "rollback": return followedCommand(ctx, "rollback", rest, stdout, stderr) + case "converge": + return followedCommand(ctx, "converge", rest, stdout, stderr) case "history": err = cmdHistory(ctx, rest, stdout) case "help", "-h", "--help": @@ -86,10 +88,14 @@ func usage(w io.Writer) { snowdeploy status snowdeploy deploy [--digest sha256:...] snowdeploy rollback [--to sha256:...] + snowdeploy converge snowdeploy history [-n 20] snowdeploy validate [--volume-prefixes p1,p2] [--env-file-prefixes p1,p2] -Connection flags (status, deploy, rollback, history): +converge re-applies the merged manifest without moving the image pin — for an +env, volume, or template edit that merged on its own. + +Connection flags (status, deploy, rollback, converge, history): --server daemon base URL (env `+envServer+`) --token-file file holding the CLI bearer token (env `+envTokenFile+`) `) @@ -180,11 +186,16 @@ func followedCommand( fs := newFlagSet(action, stderr) server, tokenFile := connectionFlags(fs) - digestFlag := "digest" - if action == "rollback" { - digestFlag = "to" + // converge has no digest to accept: it re-applies the pin main already + // holds, so it registers no digest flag at all. + digest := new(string) + if action != "converge" { + digestFlag := "digest" + if action == "rollback" { + digestFlag = "to" + } + digest = fs.String(digestFlag, "", "digest to pin (sha256:...)") } - digest := fs.String(digestFlag, "", "digest to pin (sha256:...)") timeout := fs.Duration("timeout", defaultFollowTimeout, "give up waiting for a terminal state after this long") diff --git a/docs/superpowers/specs/2026-08-27-converge-action-design.md b/docs/superpowers/specs/2026-08-27-converge-action-design.md new file mode 100644 index 0000000..b15a63e --- /dev/null +++ b/docs/superpowers/specs/2026-08-27-converge-action-design.md @@ -0,0 +1,133 @@ +# Converge action + +2026-08-27. Motivated by the snowsys Phase 12 budget red test: a manifest +change that only edited `env` (a new `BUDGET_MONTHLY_USD`) merged to main, but +`Engine.start()` refuses any request whose digest equals the manifest pin +(`ErrAlreadyAtDigest`), so the rendered unit on the host kept the old +environment until the next digest change. Observed live 2026-08-27. + +## Problem + +The engine's only two actions — deploy and rollback — are both "pin a digest". +A manifest edit that changes env, volumes, or the template without moving the +digest has no path to convergence: nothing re-renders the unit from the merged +manifest. + +## Approaches considered + +1. **Explicit `converge` action (chosen).** A third engine action that applies + the merged manifest as main has it right now: sync, render, write the unit, + restart, probe. No pull request — the change it applies already merged + through the configuration repository's own review. Distinct intent, small + surface, honest journal receipts. +2. **Same-digest deploys fall through to apply when the rendered unit would + differ.** Requires reading the current unit back from the host (a new + `Systemd` surface) and comparing renders, and it muddies the deploy + contract ("pin a digest") with a second meaning. Rejected. +3. **Automatic unit-content drift reconciliation.** A watcher that re-renders + every manifest and applies on difference. A standing-behavior change with + its own failure modes; far beyond what the incident needs. Rejected. + +## Design + +### Engine + +- `journal.ActionConverge = "converge"`. +- `Engine.Converge(ctx, service, actor) (id int64, joined bool, err error)`, + wired through the existing `start()`: the resolve function returns the + current pin, and the `ErrAlreadyAtDigest` guard is skipped for converge + (equality is the point). `setDigest` is skipped; there is no content change + to propose. +- Deduplication: `pendingRun` gains the action. A second converge click joins + the pending converge run; it never joins a deploy that happens to carry the + same digest, because a deploy's apply may predate the manifest edit the + converge was clicked for. +- The run reuses `run()`'s prologue — detected event, sync, per-service lock + (the one-run-per-service queue) — then branches into a converge path that + never touches the merge queue (`mergeMu`): converge opens no pull request, + so it has no open-to-merge span to serialize. +- At the front of the queue the converge **syncs again** and then re-reads the + manifest. The prologue sync predates the queue wait, so the mirror can be + stale by the time the converge's turn comes — or still pinning a digest a + revert since took back. The sync failure wording goes through the same + `syncFailure` path as every other sync, so the credential-bearing transport + error never reaches the journal. +- If interleaved deploys moved the pin while the converge waited, converge + applies the new pin — host matches main is the whole contract — and rewrites + the still-unfinished journal entry's digests to the pin actually applied + (`Journal.SetDigests`, which refuses finished receipts). Receipts stay + immutable; an in-flight entry is not yet a receipt. Without the rewrite, a + healthy converge row carrying the click-time pin would feed + `PreviousHealthyDigest` a digest that was never verified by this run. +- States: detected → reconciling → probing → healthy/failed. The UI's + progress rail already renders runs that skip pr-open/checks/merged (the + same-digest deploy path produces the same shape). +- `applyMerged` is split: the template-fetch-and-apply tail becomes + `applyManifest(ctx, req, m)`, shared by the deploy path and converge. + Converge does not use `applyMerged` itself — its digest assertion is against + the Begin-time pin, which converge deliberately does not hold fixed. + +### No automatic rollback + +Deploy can roll back because it holds the click-time manifest (`originalRaw`). +For converge, the click-time manifest *is* the merged manifest being applied; +the previously rendered unit's content is recorded nowhere. A converge whose +probe fails finishes `failed`, with a detail saying exactly that: there is no +previous unit to restore — fix the manifest and converge again, or deploy a +known-good digest. `LastHealthyDigest` is untouched by a failed run, so a +subsequent bare rollback still resolves sensibly. + +A healthy converge journals `new_digest` = the current pin with state healthy, +which is true and keeps the healthy-digest queries correct. + +### API + +`POST /api/v1/services/{name}/converge`, authenticated like deploy and +rollback, no request body (any body is ignored — there is no digest to accept, +and accepting one would invite the same-digest confusion converge exists to +end). Responds 202 with `{journalId, joined}` via the shared `dispatch`. + +### CLI + +`snowdeploy converge ` — the followed-command flow (subscribe to +events first, POST, stream to a terminal state, exit nonzero unless healthy), +minus the digest flag and the latest-digest resolution. `client.startDeploy` +sends no body for converge. + +### Out of scope + +- A converge button in the web UI. The UI renders converge runs arriving on + the event stream (action rides on every event and history row); adding a + button is a separate, purely frontend change. +- Skipping the restart when the rendered unit is byte-identical to what is on + the host. Converge is a deliberate operator action; an occasional no-op + restart, gated by the probe, is acceptable. Revisit only if converge grows + an automated caller. + +## Tests + +Engine (`internal/deploy/engine_test.go` harness): + +- Converge applies the current manifest with no PR opened, states + detected→reconciling→probing→healthy, journal receipt action=converge. +- Converge applies the manifest as it stands at the front of the queue after + an interleaved deploy moved it, and its receipt records the applied pin. +- Converge syncs the mirror at the front of the queue: an edit that lands on + main after the prologue sync is still applied (the fake repo keeps main and + the fetched mirror apart so staleness is visible at all). +- A front-of-queue sync failure fails the run in the repository's own words, + never the transport error. +- Converge probe failure finishes failed, applies nothing else, opens no PR, + and leaves LastHealthyDigest unchanged. +- A duplicate converge click joins the pending converge run. +- A converge does not join a pending deploy carrying the same digest. +- Sync failure fails the run without leaking the transport error (existing + wording path). + +API (`internal/api/api_test.go`): + +- POST converge dispatches to the engine and returns 202 with the journal id; + unknown service 404s; unauthenticated 401s (existing middleware table). + +CLI (`cmd/snowdeploy/client_test.go` / main): converge posts to the right +path with no body and follows to a terminal state. diff --git a/internal/api/api.go b/internal/api/api.go index bf0d68e..f84ca8a 100644 --- a/internal/api/api.go +++ b/internal/api/api.go @@ -49,6 +49,7 @@ var digestPattern = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) type Engine interface { Deploy(ctx context.Context, service, digest, actor string) (int64, bool, error) Rollback(ctx context.Context, service, toDigest, actor string) (int64, bool, error) + Converge(ctx context.Context, service, actor string) (int64, bool, error) Drift(ctx context.Context) (map[string]string, error) QueueDepth() int64 } @@ -185,6 +186,7 @@ func (s *Server) Handler() http.Handler { mux.Handle("GET /api/v1/services/{name}/revisions", s.authed(s.handleRevisions)) mux.Handle("POST /api/v1/services/{name}/deploy", s.authed(s.handleDeploy)) mux.Handle("POST /api/v1/services/{name}/rollback", s.authed(s.handleRollback)) + mux.Handle("POST /api/v1/services/{name}/converge", s.authed(s.handleConverge)) mux.Handle("GET /api/v1/events", s.authed(func(w http.ResponseWriter, r *http.Request, _ string) { s.handleEvents(w, r) })) @@ -424,6 +426,19 @@ func (s *Server) handleRollback(w http.ResponseWriter, r *http.Request, actor st }) } +// handleConverge re-applies the merged manifest as it stands. It reads no +// body: converge has no digest to accept, and accepting one would reintroduce +// the same-digest ambiguity the action exists to end. +func (s *Server) handleConverge(w http.ResponseWriter, r *http.Request, actor string) { + name, ok := s.service(w, r) + if !ok { + return + } + s.dispatch(w, r, func() (int64, bool, error) { + return s.opts.Engine.Converge(r.Context(), name, actor) + }) +} + // accepted is the body of a 202: the journal id to follow, and whether the // click joined a run already carrying the same digest instead of starting one. type accepted struct { diff --git a/internal/api/api_test.go b/internal/api/api_test.go index b09e59e..670f9a3 100644 --- a/internal/api/api_test.go +++ b/internal/api/api_test.go @@ -68,6 +68,16 @@ func (e *fakeEngine) Rollback(_ context.Context, service, digest, actor string) return int64(len(e.calls)), e.joined, nil } +func (e *fakeEngine) Converge(_ context.Context, service, actor string) (int64, bool, error) { + e.mu.Lock() + defer e.mu.Unlock() + e.calls = append(e.calls, deployCall{service, "", actor, "converge"}) + if e.err != nil { + return 0, false, e.err + } + return int64(len(e.calls)), e.joined, nil +} + func (e *fakeEngine) Drift(context.Context) (map[string]string, error) { e.mu.Lock() defer e.mu.Unlock() @@ -614,6 +624,40 @@ func TestRollbackWithoutDigestIsAllowed(t *testing.T) { } } +func TestConvergeTakesNoBodyAndDispatches(t *testing.T) { + h := newHarness(t) + resp := h.do(t, http.MethodPost, "/api/v1/services/web/converge", ``, remoteUser("admin")) + if resp.StatusCode != http.StatusAccepted { + t.Fatalf("converge = %d, want 202", resp.StatusCode) + } + var body struct { + JournalID int64 `json:"journalId"` + Joined bool `json:"joined"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err != nil { + t.Fatalf("decode: %v", err) + } + if body.JournalID == 0 { + t.Error("no journal id returned") + } + calls := h.engine.recorded() + if len(calls) != 1 || calls[0].Action != "converge" || + calls[0].Service != "web" || calls[0].Digest != "" { + t.Errorf("engine call = %+v", calls) + } + if calls[0].Actor != "admin" { + t.Errorf("actor = %q", calls[0].Actor) + } +} + +func TestConvergeOnAnUnknownServiceIsNotFound(t *testing.T) { + h := newHarness(t) + resp := h.do(t, http.MethodPost, "/api/v1/services/nope/converge", ``, remoteUser("admin")) + if resp.StatusCode != http.StatusNotFound { + t.Fatalf("converge on unknown service = %d, want 404", resp.StatusCode) + } +} + func TestEngineErrorSurfacesAsBadRequest(t *testing.T) { h := newHarness(t) h.engine.err = errors.New("service is already pinned to that digest") diff --git a/internal/deploy/engine.go b/internal/deploy/engine.go index 08ecf5b..a094de2 100644 --- a/internal/deploy/engine.go +++ b/internal/deploy/engine.go @@ -46,6 +46,7 @@ type Store interface { Progress(id int64, state, detail string) error Finish(id int64, state, detail string) error SetPR(id int64, prNumber int) error + SetDigests(id int64, oldDigest, newDigest string) error SetMergeSHA(id int64, sha string) error LastHealthyDigest(service string) (string, error) PreviousHealthyDigest(service, notDigest string) (string, error) @@ -177,6 +178,17 @@ func (e *Engine) Rollback( }) } +// Converge applies the manifest as main has it right now — same digest, fresh +// render — for the change a deploy cannot carry: an env, volume, or template +// edit that merged without moving the image pin. It opens no pull request; the +// change it applies already went through the configuration repository's review. +func (e *Engine) Converge( + _ context.Context, service, actor string, +) (id int64, joined bool, err error) { + return e.start(journal.ActionConverge, service, actor, + func(current string) (string, error) { return current, nil }) +} + // resolveTarget turns a request into the digest to pin, given what is pinned. type resolveTarget func(currentDigest string) (string, error) @@ -200,13 +212,18 @@ func (e *Engine) start( if !digestPattern.MatchString(digest) { return 0, false, fmt.Errorf("%q is not a sha256 digest pin", digest) } - if current.Image.Digest == digest { - return 0, false, fmt.Errorf("%s: %w (%s)", service, ErrAlreadyAtDigest, digest) - } - - newContent, err := setDigest(raw, current.Image.Digest, digest) - if err != nil { - return 0, false, err + // For converge the pinned digest is the point, and there is no content + // change to propose: the run re-renders and re-applies what main already + // holds. + newContent := raw + if action != journal.ActionConverge { + if current.Image.Digest == digest { + return 0, false, fmt.Errorf("%s: %w (%s)", service, ErrAlreadyAtDigest, digest) + } + newContent, err = setDigest(raw, current.Image.Digest, digest) + if err != nil { + return 0, false, err + } } // A second click for a target already queued or running is the same @@ -218,7 +235,13 @@ func (e *Engine) start( // fast local SQLite write, cheap enough to hold the lock across. e.mu.Lock() pending, live := e.pending[service] - if live && pending.digest == digest { + // A converge only ever joins a converge: a pending deploy at the same + // digest froze its manifest before the edit the converge was clicked for, + // so answering with that run could leave the edit unapplied. Deploys and + // rollbacks keep joining each other — both mean "pin this digest". + sameKind := (pending.action == journal.ActionConverge) == + (action == journal.ActionConverge) + if live && sameKind && pending.digest == digest { e.mu.Unlock() return pending.id, true, nil } @@ -239,7 +262,7 @@ func (e *Engine) start( // dedup point for every click still matching it. This run proceeds to the // queue unregistered and simply offers no join point of its own. if !live { - e.pending[service] = pendingRun{digest: digest, id: id} + e.pending[service] = pendingRun{action: action, digest: digest, id: id} } e.mu.Unlock() @@ -266,8 +289,9 @@ func (e *Engine) start( return id, false, nil } -// pendingRun is a deploy that has a journal entry but has not yet finished. +// pendingRun is a run that has a journal entry but has not yet finished. type pendingRun struct { + action string digest string id int64 } @@ -306,10 +330,14 @@ func (e *Engine) run(req runRequest) { return } - e.emit(req, StateDetected, "queued for the merge queue") + queuedDetail := "queued for the merge queue" + if req.action == journal.ActionConverge { + queuedDetail = "queued for the service's turn" + } + e.emit(req, StateDetected, queuedDetail) release, err := e.lock(ctx, req.service) if err != nil { - detail := fmt.Sprintf("queued deploy abandoned: %v", err) + detail := fmt.Sprintf("queued %s abandoned: %v", req.action, err) if errors.Is(err, context.Canceled) { detail += "; the daemon was restarting; re-run" } @@ -318,6 +346,13 @@ func (e *Engine) run(req runRequest) { } defer release() + // A converge proposes nothing, so it has no business in the merge queue: + // it re-reads main at the front of its own service's queue and applies it. + if req.action == journal.ActionConverge { + e.converge(ctx, req) + return + } + e.lockMergeQueue() // The manifest this run froze at click time is stale by the time it @@ -486,6 +521,49 @@ func (e *Engine) syncFailure(err error) string { return detail } +// converge applies the manifest as main holds it at the front of the queue. +// The click-time sync predates the queue wait, so it syncs again here — the +// mirror can be stale by minutes, or still pinning a digest a revert since +// took back. Interleaved deploys may also have moved the pin while this run +// waited; applying the file as it is now is the contract, and the receipt is +// rewritten to the pin actually applied before it can finish. +// +// There is no rollback from here: the unit the host ran before this converge +// is recorded nowhere, so a failed probe finishes failed and says so. +func (e *Engine) converge(ctx context.Context, req runRequest) { + if _, err := e.opts.Repo.Sync(ctx); err != nil { + e.fail(req, "sync merged state: "+e.syncFailure(err)) + return + } + m, _, err := e.opts.Repo.Manifest(req.service) + if err != nil { + e.fail(req, fmt.Sprintf( + "could not re-read the manifest at the front of the queue: %v", err)) + return + } + if m.Image.Digest != req.newDigest { + if err := e.opts.Journal.SetDigests( + req.id, m.Image.Digest, m.Image.Digest); err != nil { + e.fail(req, fmt.Sprintf( + "the pin moved to %s while queued and the journal could not "+ + "record it: %v", m.Image.Digest, err)) + return + } + } + req.oldDigest, req.newDigest = m.Image.Digest, m.Image.Digest + e.emit(req, StateReconciling, fmt.Sprintf( + "re-rendering the merged manifest, still pinned to %s", + shortDigest(m.Image.Digest))) + if err := e.applyManifest(ctx, req, m); err != nil { + e.fail(req, fmt.Sprintf( + "%v; a converge has no previous unit to restore — fix the manifest "+ + "and converge again, or deploy a known-good digest", err)) + return + } + e.finish(req, StateHealthy, fmt.Sprintf( + "%s converged; healthy on %s", req.service, req.newDigest)) +} + // applyMerged pulls the merged branch and walks it onto the host. func (e *Engine) applyMerged(ctx context.Context, req runRequest) error { if _, err := e.opts.Repo.Sync(ctx); err != nil { @@ -499,6 +577,13 @@ func (e *Engine) applyMerged(ctx context.Context, req runRequest) error { return fmt.Errorf( "merged manifest pins %s, not the requested %s", m.Image.Digest, req.newDigest) } + return e.applyManifest(ctx, req, m) +} + +// applyManifest walks one manifest onto the host, streaming the phases. +func (e *Engine) applyManifest( + ctx context.Context, req runRequest, m *manifest.Manifest, +) error { tmpl, err := e.opts.Repo.Template(m.Template) if err != nil { return err diff --git a/internal/deploy/engine_test.go b/internal/deploy/engine_test.go index 007b986..1d97080 100644 --- a/internal/deploy/engine_test.go +++ b/internal/deploy/engine_test.go @@ -40,27 +40,46 @@ health: // ---- fakes ----------------------------------------------------------------- +// fakeRepo keeps the configuration repository's main and the daemon's fetched +// mirror apart, the way the real gitops repo does: Manifest reads the mirror, +// merge lands on main, and only Sync brings the mirror up to date. A fake +// with a single view would make every staleness bug invisible. type fakeRepo struct { mu sync.Mutex - raw map[string]string + main map[string]string + mirror map[string]string templates map[string]string syncs int syncErr error + + // onSync runs after each successful sync, outside the lock, with the + // sync's ordinal — a test's hook for moving main between two syncs. + onSync func(syncs int) } func newFakeRepo(digest string) *fakeRepo { return &fakeRepo{ - raw: map[string]string{"web": manifestYAML(digest)}, + main: map[string]string{"web": manifestYAML(digest)}, + mirror: map[string]string{"web": manifestYAML(digest)}, templates: map[string]string{"web.container.tmpl": "[Container]\nImage={{.Image.Repository}}@{{.Image.Digest}}\n"}, } } func (r *fakeRepo) Sync(context.Context) (string, error) { r.mu.Lock() - defer r.mu.Unlock() r.syncs++ if r.syncErr != nil { - return "", r.syncErr + err := r.syncErr + r.mu.Unlock() + return "", err + } + for service, content := range r.main { + r.mirror[service] = content + } + n, onSync := r.syncs, r.onSync + r.mu.Unlock() + if onSync != nil { + onSync(n) } return "headsha", nil } @@ -79,7 +98,7 @@ func (r *fakeRepo) SyncState() (bool, error) { func (r *fakeRepo) Manifest(service string) (*manifest.Manifest, []byte, error) { r.mu.Lock() defer r.mu.Unlock() - raw, ok := r.raw[service] + raw, ok := r.mirror[service] if !ok { return nil, nil, fmt.Errorf("no manifest for %s", service) } @@ -104,17 +123,34 @@ func (r *fakeRepo) Services() ([]string, error) { r.mu.Lock() defer r.mu.Unlock() var out []string - for name := range r.raw { + for name := range r.mirror { out = append(out, name) } return out, nil } -// merge simulates a PR landing on main. +// seed plants a service in both main and the mirror, as if it had always +// been there. +func (r *fakeRepo) seed(service, content string) { + r.mu.Lock() + defer r.mu.Unlock() + r.main[service] = content + r.mirror[service] = content +} + +// setOnSync installs the after-sync hook. +func (r *fakeRepo) setOnSync(fn func(syncs int)) { + r.mu.Lock() + defer r.mu.Unlock() + r.onSync = fn +} + +// merge simulates a PR landing on main. The mirror learns of it only when +// something syncs. func (r *fakeRepo) merge(service string, content []byte) { r.mu.Lock() defer r.mu.Unlock() - r.raw[service] = string(content) + r.main[service] = string(content) } type prCall struct { @@ -217,11 +253,12 @@ func (p *fakePR) snapshot() ([]prCall, []int, []int) { } type fakeApplier struct { - mu sync.Mutex - applied []string - appliedBy []string - failOn map[string]error - gate chan struct{} + mu sync.Mutex + applied []string + appliedBy []string + appliedEnv []map[string]string + failOn map[string]error + gate chan struct{} } func (a *fakeApplier) Apply( @@ -230,6 +267,7 @@ func (a *fakeApplier) Apply( a.mu.Lock() a.applied = append(a.applied, m.Image.Digest) a.appliedBy = append(a.appliedBy, m.Name) + a.appliedEnv = append(a.appliedEnv, m.Env) gate := a.gate err := a.failOn[m.Image.Digest] a.mu.Unlock() @@ -252,6 +290,13 @@ func (a *fakeApplier) calls() []string { return append([]string(nil), a.applied...) } +// envs is each applied manifest's env map, in order. +func (a *fakeApplier) envs() []map[string]string { + a.mu.Lock() + defer a.mu.Unlock() + return append([]map[string]string(nil), a.appliedEnv...) +} + // callsFor is the digests applied for one service, in order. func (a *fakeApplier) callsFor(name string) []string { a.mu.Lock() @@ -993,9 +1038,7 @@ func TestExhaustedMergeRetriesFailAndCloseThePR(t *testing.T) { // protection strands whichever merges second. func TestConcurrentDeploysSerializeTheMergeSpan(t *testing.T) { h := newHarness(t, digestA) - h.repo.mu.Lock() - h.repo.raw["api"] = strings.Replace(manifestYAML(digestA), "name: web", "name: api", 1) - h.repo.mu.Unlock() + h.repo.seed("api", strings.Replace(manifestYAML(digestA), "name: web", "name: api", 1)) h.insp.mu.Lock() h.insp.running["api"] = digestA h.insp.mu.Unlock() @@ -1192,9 +1235,7 @@ func TestSyncFailureFailsTheRunWithoutLeakingTheError(t *testing.T) { // wait. It returns web's journal id after both runs finished. func landedWhileQueued(t *testing.T, h *harness) int64 { t.Helper() - h.repo.mu.Lock() - h.repo.raw["api"] = strings.Replace(manifestYAML(digestA), "name: web", "name: api", 1) - h.repo.mu.Unlock() + h.repo.seed("api", strings.Replace(manifestYAML(digestA), "name: web", "name: api", 1)) h.insp.mu.Lock() h.insp.running["api"] = digestA h.insp.mu.Unlock() @@ -1279,3 +1320,266 @@ func TestQueuedDeployWhoseDigestLandedButIsNotRunningApplies(t *testing.T) { } } } + +// ---- converge -------------------------------------------------------------- + +// manifestYAMLWithEnv is a manifest whose only difference from +// manifestYAML(digest) is an env block — the shape change converge exists for. +func manifestYAMLWithEnv(digest string) string { + return manifestYAML(digest) + "env:\n BUDGET_MONTHLY_USD: \"25\"\n" +} + +func TestConvergeAppliesTheMergedManifestWithoutAPullRequest(t *testing.T) { + h := newHarness(t, digestA) + h.repo.merge("web", []byte(manifestYAMLWithEnv(digestA))) + + id, joined, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + if joined { + t.Error("the first converge click reported itself as joining a run") + } + h.waitTerminal(t) + + want := []string{ + StateDetected, StateDetected, StateReconciling, StateReconciling, + StateProbing, StateHealthy, + } + if got := h.states(); len(got) != len(want) { + t.Fatalf("states = %v, want %v", got, want) + } else { + for i := range want { + if got[i] != want[i] { + t.Fatalf("state %d = %q, want %q (full: %v)", i, got[i], want[i], got) + } + } + } + + opened, _, _ := h.pr.snapshot() + if len(opened) != 0 { + t.Errorf("converge opened a pull request: %+v", opened) + } + if applied := h.applier.calls(); len(applied) != 1 || applied[0] != digestA { + t.Fatalf("applied = %v, want the pinned digest once", applied) + } + if envs := h.applier.envs(); len(envs) != 1 || envs[0]["BUDGET_MONTHLY_USD"] != "25" { + t.Errorf("the applied manifest lost the merged env change: %v", envs) + } + + e := h.entry(t, id) + if e.Action != journal.ActionConverge { + t.Errorf("action = %q, want converge", e.Action) + } + if e.State != journal.StateHealthy { + t.Errorf("journal state = %q (%s), want healthy", e.State, e.Detail) + } + if e.OldDigest != digestA || e.NewDigest != digestA { + t.Errorf("journal digests = %s -> %s, want the unchanged pin", e.OldDigest, e.NewDigest) + } + if e.Actor != "admin" { + t.Errorf("actor = %q", e.Actor) + } +} + +// A converge clicked while a deploy's proposal is still open journals the +// click-time pin, but by its turn the pin has moved. It must apply — and its +// receipt must record — the pin main holds at the front of the queue. +func TestConvergeAppliesTheManifestAtTheFrontOfTheQueue(t *testing.T) { + h := newHarness(t, digestA) + + gate := make(chan struct{}) + var reached sync.Once + firstOpen := make(chan struct{}) + h.pr.mu.Lock() + h.pr.onOpen = func(service string, content []byte) { + reached.Do(func() { close(firstOpen) }) + <-gate + h.repo.merge(service, content) + } + h.pr.mu.Unlock() + + if _, _, err := h.engine.Deploy(t.Context(), "web", digestB, "admin"); err != nil { + t.Fatalf("Deploy: %v", err) + } + <-firstOpen + + id, _, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + close(gate) + h.engine.Wait() + + if applied := h.applier.calls(); len(applied) != 2 || applied[1] != digestB { + t.Fatalf("applied = %v, want the converge to apply the moved pin %s", + applied, digestB) + } + e := h.entry(t, id) + if e.State != journal.StateHealthy { + t.Fatalf("converge state = %q (%s), want healthy", e.State, e.Detail) + } + if e.OldDigest != digestB || e.NewDigest != digestB { + t.Errorf("receipt digests = %s -> %s, want the applied pin %s twice: "+ + "a stale receipt feeds the healthy-digest queries the wrong pin", + e.OldDigest, e.NewDigest, digestB) + } +} + +// The click-time sync happens before the queue wait, so main can move between +// it and the converge's turn. The converge must sync again at the front of +// the queue, or it applies whatever the mirror last happened to hold. +func TestConvergeSyncsTheMirrorAtTheFrontOfTheQueue(t *testing.T) { + h := newHarness(t, digestA) + h.repo.setOnSync(func(syncs int) { + if syncs == 1 { + h.repo.merge("web", []byte(manifestYAMLWithEnv(digestA))) + } + }) + + if _, _, err := h.engine.Converge(t.Context(), "web", "admin"); err != nil { + t.Fatalf("Converge: %v", err) + } + h.waitTerminal(t) + + envs := h.applier.envs() + if len(envs) != 1 || envs[0]["BUDGET_MONTHLY_USD"] != "25" { + t.Fatalf("applied env = %v: the converge applied the stale mirror "+ + "instead of syncing at the front of the queue", envs) + } +} + +// A sync failure at the front of the queue must fail the run in the +// repository's own words: the transport error can quote a credential-bearing +// remote URL, and the journal detail reaches every client. +func TestConvergeFrontOfQueueSyncFailureDoesNotLeak(t *testing.T) { + h := newHarness(t, digestA) + transport := "clone https://x-access-token:ghs_synthetic_never_real@github.com/acme/config.git: auth failed" + h.repo.setOnSync(func(syncs int) { + if syncs == 1 { + h.repo.mu.Lock() + h.repo.syncErr = errors.New(transport) + h.repo.mu.Unlock() + } + }) + + id, _, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + h.waitTerminal(t) + + e := h.entry(t, id) + if e.State != journal.StateFailed { + t.Fatalf("state = %q (%s), want failed", e.State, e.Detail) + } + if strings.Contains(e.Detail, "ghs_synthetic_never_real") { + t.Fatalf("the transport error reached the journal: %q", e.Detail) + } + if !strings.Contains(e.Detail, "could not be fetched") { + t.Errorf("detail lost the repository's own reason: %q", e.Detail) + } + if applied := h.applier.calls(); len(applied) != 0 { + t.Errorf("a converge that could not sync still applied: %v", applied) + } +} + +func TestConvergeProbeFailureFailsWithoutRollback(t *testing.T) { + h := newHarness(t, digestA) + h.applier.failOn[digestA] = errors.New("probe: connection refused") + + id, _, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + h.waitTerminal(t) + + e := h.entry(t, id) + if e.State != journal.StateFailed { + t.Fatalf("state = %q (%s), want failed", e.State, e.Detail) + } + if !strings.Contains(e.Detail, "no previous unit") { + t.Errorf("detail does not warn that nothing can be restored: %q", e.Detail) + } + if applied := h.applier.calls(); len(applied) != 1 { + t.Errorf("a failed converge applied again: %v", applied) + } + opened, _, _ := h.pr.snapshot() + if len(opened) != 0 { + t.Errorf("a failed converge opened a pull request: %+v", opened) + } + if _, err := h.jrnl.LastHealthyDigest("web"); !errors.Is(err, journal.ErrNoHealthyDeploy) { + t.Errorf("a failed converge minted a healthy digest: %v", err) + } +} + +func TestDuplicateConvergeClickJoinsThePendingConverge(t *testing.T) { + h := newHarness(t, digestA) + h.applier.gate = make(chan struct{}) + + first, firstJoined, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + second, secondJoined, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("second Converge: %v", err) + } + close(h.applier.gate) + h.engine.Wait() + + if first != second { + t.Fatalf("duplicate converge journaled a second run: %d vs %d", first, second) + } + if firstJoined || !secondJoined { + t.Errorf("join flags = %v, %v; want the second click to join the first", + firstJoined, secondJoined) + } + entries, err := h.jrnl.Recent("web", 50) + if err != nil { + t.Fatalf("Recent: %v", err) + } + if len(entries) != 1 { + t.Fatalf("duplicate converge journaled %d runs, want 1", len(entries)) + } +} + +// A deploy whose proposal has merged still holds the pending entry for its +// digest, which by then equals the pin. A converge clicked in that window is +// asking for a fresher apply than that run will perform, so it must start its +// own run rather than join the deploy's. +func TestConvergeDoesNotJoinAPendingDeployAtTheSameDigest(t *testing.T) { + h := newHarness(t, digestA) + gate := make(chan struct{}) + h.applier.gate = gate + + first, _, err := h.engine.Deploy(t.Context(), "web", digestB, "admin") + if err != nil { + t.Fatalf("Deploy: %v", err) + } + deadline := time.Now().Add(2 * time.Second) + for len(h.applier.calls()) == 0 { + if time.Now().After(deadline) { + t.Fatal("the deploy never reached Apply") + } + time.Sleep(time.Millisecond) + } + + converge, joined, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + if joined || converge == first { + t.Fatalf("converge joined the pending deploy: id=%d joined=%v", converge, joined) + } + + close(gate) + h.applier.mu.Lock() + h.applier.gate = nil + h.applier.mu.Unlock() + h.engine.Wait() + + if e := h.entry(t, converge); e.State != journal.StateHealthy { + t.Errorf("converge state = %q (%s), want healthy", e.State, e.Detail) + } +} diff --git a/internal/deploy/states.go b/internal/deploy/states.go index c590af5..70006cf 100644 --- a/internal/deploy/states.go +++ b/internal/deploy/states.go @@ -1,7 +1,8 @@ // Package deploy is the merge-first state machine: it proposes a manifest // change, waits for the configuration repository's own checks, merges, applies // the merged state to the host, probes, and journals — rolling back and -// reverting main when the probe fails. +// reverting main when the probe fails. Converge is the one action that skips +// the proposal: it re-applies what main already holds. package deploy import "github.com/SnowballSH/snowdeploy/internal/journal" diff --git a/internal/journal/journal.go b/internal/journal/journal.go index d983763..5a64105 100644 --- a/internal/journal/journal.go +++ b/internal/journal/journal.go @@ -16,6 +16,7 @@ import ( const ( ActionDeploy = "deploy" ActionRollback = "rollback" + ActionConverge = "converge" ActionAutoRollback = "auto-rollback" ) @@ -149,6 +150,16 @@ func (j *Journal) SetPR(id int64, prNumber int) error { prNumber, id) } +// SetDigests rewrites an in-flight entry's digests, for the run whose target +// is settled only at the front of the queue: a converge records the pin it +// actually applied, not the one main held at click time. +func (j *Journal) SetDigests(id int64, oldDigest, newDigest string) error { + return j.update(id, ` + UPDATE deploys SET old_digest = ?, new_digest = ? + WHERE id = ? AND finished_at IS NULL`, + oldDigest, newDigest, id) +} + // SetMergeSHA records the merge commit an in-flight entry produced. func (j *Journal) SetMergeSHA(id int64, sha string) error { return j.update(id, ` diff --git a/internal/journal/journal_test.go b/internal/journal/journal_test.go index 409e4e1..fa3216e 100644 --- a/internal/journal/journal_test.go +++ b/internal/journal/journal_test.go @@ -318,3 +318,36 @@ func TestOpenIsIdempotentAcrossRestart(t *testing.T) { t.Errorf("journal did not survive restart: %+v", got) } } + +func TestSetDigestsRewritesAnUnfinishedEntry(t *testing.T) { + j := open(t) + id, err := j.Begin(sample()) + if err != nil { + t.Fatalf("Begin: %v", err) + } + if err := j.SetDigests(id, "sha256:ccc", "sha256:ccc"); err != nil { + t.Fatalf("SetDigests: %v", err) + } + got, err := j.Recent("web", 10) + if err != nil { + t.Fatalf("Recent: %v", err) + } + if got[0].OldDigest != "sha256:ccc" || got[0].NewDigest != "sha256:ccc" { + t.Errorf("digests = %s -> %s, want the rewritten pin twice", + got[0].OldDigest, got[0].NewDigest) + } +} + +func TestSetDigestsRefusesAFinishedReceipt(t *testing.T) { + j := open(t) + id, err := j.Begin(sample()) + if err != nil { + t.Fatalf("Begin: %v", err) + } + if err := j.Finish(id, StateHealthy, "done"); err != nil { + t.Fatalf("Finish: %v", err) + } + if err := j.SetDigests(id, "sha256:ccc", "sha256:ccc"); !errors.Is(err, ErrAlreadyFinished) { + t.Fatalf("SetDigests on a receipt = %v, want ErrAlreadyFinished", err) + } +}