From b49482f6634f14e469371d839f31045b3f89edbd Mon Sep 17 00:00:00 2001 From: SnowballSH Date: Thu, 27 Aug 2026 09:10:52 -0400 Subject: [PATCH 1/4] Design the converge action: apply the merged manifest when only its shape changed --- .../2026-08-27-converge-action-design.md | 120 ++++++++++++++++++ 1 file changed, 120 insertions(+) create mode 100644 docs/superpowers/specs/2026-08-27-converge-action-design.md 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..06c1e42 --- /dev/null +++ b/docs/superpowers/specs/2026-08-27-converge-action-design.md @@ -0,0 +1,120 @@ +# 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 manifest is re-read and applied as it is now. + If interleaved deploys moved the pin while the converge waited, converge + applies the new pin — host matches main is the whole contract. The journal + entry keeps its Begin-time digests (receipts are immutable); the emitted + details name the digest actually applied. +- 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. +- 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. From aba86063d61484f3c0ca94bd61eeff96b04758f7 Mon Sep 17 00:00:00 2001 From: SnowballSH Date: Thu, 27 Aug 2026 09:17:51 -0400 Subject: [PATCH 2/4] Add converge: re-apply the merged manifest when only its shape changed MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A manifest edit that changes env, volumes, or the template without moving the image digest had no path to the host: Deploy refuses the pinned digest (ErrAlreadyAtDigest), so the rendered unit kept the old environment until the next digest change — observed live 2026-08-27 during the snowsys Phase 12 budget red test. Converge is a third engine action that applies the manifest as main holds it at the front of the service's queue: sync, re-render, write the unit, restart, probe. It opens no pull request (the change it applies already merged through review) and never enters the merge queue. It journals a run like any other, joins duplicate converge clicks, and deliberately never joins a pending deploy at the same digest, whose frozen manifest may predate the edit. There is no automatic rollback: the previously rendered unit is recorded nowhere, so a failed probe finishes failed and says exactly that. Exposed as POST /api/v1/services/{name}/converge (no body) and 'snowdeploy converge ', which follows the event stream like deploy and rollback. --- README.md | 16 ++- cmd/snowdeploy/client.go | 5 +- cmd/snowdeploy/client_test.go | 59 +++++++++ cmd/snowdeploy/main.go | 21 +++- internal/api/api.go | 15 +++ internal/api/api_test.go | 44 +++++++ internal/deploy/engine.go | 93 ++++++++++++-- internal/deploy/engine_test.go | 215 ++++++++++++++++++++++++++++++++- internal/deploy/states.go | 3 +- internal/journal/journal.go | 1 + 10 files changed, 445 insertions(+), 27 deletions(-) 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/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..4819d51 100644 --- a/internal/deploy/engine.go +++ b/internal/deploy/engine.go @@ -177,6 +177,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 +211,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 +234,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 +261,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 +288,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 +329,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 behind the service's in-flight run" + } + 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 +345,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 +520,34 @@ func (e *Engine) syncFailure(err error) string { return detail } +// converge applies the manifest as main holds it at the front of the queue. +// Interleaved deploys may have moved the pin while this run waited; applying +// the file as it is now is the contract, so the emitted details name the +// digest actually applied even when the receipt's Begin-time digests lag it. +// +// 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) { + 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 + } + req.newDigest = m.Image.Digest + e.emit(req, StateReconciling, fmt.Sprintf( + "re-rendering the merged manifest, digest unchanged at %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 +561,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..1d5a4e5 100644 --- a/internal/deploy/engine_test.go +++ b/internal/deploy/engine_test.go @@ -217,11 +217,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 +231,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 +254,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() @@ -1279,3 +1288,199 @@ 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) + } +} + +func TestConvergeAppliesTheManifestAtTheFrontOfTheQueue(t *testing.T) { + h := newHarness(t, digestA) + gate := make(chan struct{}) + h.applier.gate = gate + + if _, _, err := h.engine.Deploy(t.Context(), "web", digestB, "admin"); 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) + } + + id, _, err := h.engine.Converge(t.Context(), "web", "admin") + if err != nil { + t.Fatalf("Converge: %v", err) + } + close(gate) + h.applier.mu.Lock() + h.applier.gate = nil + h.applier.mu.Unlock() + 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) + } + if e := h.entry(t, id); e.State != journal.StateHealthy { + t.Errorf("converge state = %q (%s), want healthy", e.State, e.Detail) + } +} + +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..027db77 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" ) From 5f3a1c230b6bdc5c8cb788079fc93bf9cc6791eb Mon Sep 17 00:00:00 2001 From: SnowballSH Date: Thu, 27 Aug 2026 09:27:04 -0400 Subject: [PATCH 3/4] Sync at the front of the queue, and let the converge receipt tell the truth MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review caught two holes in the first cut. Converge read the mirror without syncing at the front of the queue, so it applied whatever the daemon last happened to fetch — an edit merged while it waited went unapplied yet the run finished healthy, and after a failed deploy's revert the mirror still pinned the digest that had just failed its probe. Converge now syncs before its re-read, wording failures through syncFailure so the credential-bearing transport error stays out of the journal. And when the pin moved while the converge waited, its receipt kept the click-time digests: a healthy row whose new_digest this run never verified, feeding PreviousHealthyDigest the wrong rollback target. Journal.SetDigests rewrites the still-unfinished entry to the pin actually applied; it refuses finished receipts, so receipts stay immutable. The single-view fake repo could not see either bug — merge landed straight in the manifests every read consulted. It now keeps main and the fetched mirror apart, the way the real gitops repo does, with an onSync hook for moving main between two syncs. --- .../2026-08-27-converge-action-design.md | 25 ++- internal/deploy/engine.go | 26 ++- internal/deploy/engine_test.go | 153 ++++++++++++++---- internal/journal/journal.go | 10 ++ internal/journal/journal_test.go | 33 ++++ 5 files changed, 209 insertions(+), 38 deletions(-) diff --git a/docs/superpowers/specs/2026-08-27-converge-action-design.md b/docs/superpowers/specs/2026-08-27-converge-action-design.md index 06c1e42..b15a63e 100644 --- a/docs/superpowers/specs/2026-08-27-converge-action-design.md +++ b/docs/superpowers/specs/2026-08-27-converge-action-design.md @@ -46,11 +46,19 @@ manifest. (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 manifest is re-read and applied as it is now. - If interleaved deploys moved the pin while the converge waited, converge - applies the new pin — host matches main is the whole contract. The journal - entry keeps its Begin-time digests (receipts are immutable); the emitted - details name the digest actually applied. +- 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). @@ -103,7 +111,12 @@ 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. + 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. diff --git a/internal/deploy/engine.go b/internal/deploy/engine.go index 4819d51..483177a 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) @@ -331,7 +332,7 @@ func (e *Engine) run(req runRequest) { queuedDetail := "queued for the merge queue" if req.action == journal.ActionConverge { - queuedDetail = "queued behind the service's in-flight run" + queuedDetail = "queued for the service's turn" } e.emit(req, StateDetected, queuedDetail) release, err := e.lock(ctx, req.service) @@ -521,20 +522,35 @@ func (e *Engine) syncFailure(err error) string { } // converge applies the manifest as main holds it at the front of the queue. -// Interleaved deploys may have moved the pin while this run waited; applying -// the file as it is now is the contract, so the emitted details name the -// digest actually applied even when the receipt's Begin-time digests lag it. +// 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 } - req.newDigest = m.Image.Digest + 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, digest unchanged at %s", shortDigest(m.Image.Digest))) diff --git a/internal/deploy/engine_test.go b/internal/deploy/engine_test.go index 1d5a4e5..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 { @@ -1002,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() @@ -1201,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() @@ -1350,38 +1382,105 @@ func TestConvergeAppliesTheMergedManifestWithoutAPullRequest(t *testing.T) { } } +// 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{}) - h.applier.gate = gate + 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) } - 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) - } + <-firstOpen id, _, err := h.engine.Converge(t.Context(), "web", "admin") if err != nil { t.Fatalf("Converge: %v", err) } close(gate) - h.applier.mu.Lock() - h.applier.gate = nil - h.applier.mu.Unlock() 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) } - if e := h.entry(t, id); e.State != journal.StateHealthy { - t.Errorf("converge state = %q (%s), want healthy", e.State, e.Detail) + 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) } } diff --git a/internal/journal/journal.go b/internal/journal/journal.go index 027db77..5a64105 100644 --- a/internal/journal/journal.go +++ b/internal/journal/journal.go @@ -150,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) + } +} From f2780a56cb9899d47b9e7c6b25a8936383dbebd4 Mon Sep 17 00:00:00 2001 From: SnowballSH Date: Thu, 27 Aug 2026 09:29:12 -0400 Subject: [PATCH 4/4] Say 'still pinned to', not 'unchanged': the pin may have moved since the click --- internal/deploy/engine.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/internal/deploy/engine.go b/internal/deploy/engine.go index 483177a..a094de2 100644 --- a/internal/deploy/engine.go +++ b/internal/deploy/engine.go @@ -552,7 +552,7 @@ func (e *Engine) converge(ctx context.Context, req runRequest) { } req.oldDigest, req.newDigest = m.Image.Digest, m.Image.Digest e.emit(req, StateReconciling, fmt.Sprintf( - "re-rendering the merged manifest, digest unchanged at %s", + "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(