carlos / platform Public

host: recovery answers the journal's version, keeps its staged units, and ends the run

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015QCUCpfxbCyEAVQ2m51dH5
Paul Campbell pushed by paul@keymail.dev 976025f855b72a73b6a31886236530bf62f48bcd parent e331874
2 files changed, +287 −38
  • internal/host/systemapply.go +106 −35
  • internal/host/systemapply_test.go +181 −3
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index 39def0f57..dc4757ec5 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -115,6 +115,13 @@ func productionSysOps() sysOps {
defer cancel()
out, err := exec.CommandContext(ctx, name, args...).Output()
if err != nil {
+ // Output() keeps stderr out of the returned bytes so the
+ // caller can parse them; a command that failed is the one
+ // case where what it said matters more than what it printed.
+ var ee *exec.ExitError
+ if errors.As(err, &ee) && len(ee.Stderr) > 0 {
+ return string(out), fmt.Errorf("%s %s: %w: %s", name, strings.Join(args, " "), err, strings.TrimSpace(string(ee.Stderr)))
+ }
return string(out), fmt.Errorf("%s %s: %w", name, strings.Join(args, " "), err)
}
return string(out), nil
@@ -187,12 +194,20 @@ func (a *systemApplier) Run(ctx context.Context, recover bool) error {
}
defer os.RemoveAll(a.stagingDir())
- if err := a.recoverIfNeeded(ctx); err != nil {
+ recovered, err := a.recoverIfNeeded(ctx)
+ if err != nil {
return err
}
if recover {
return nil
}
+ if recovered {
+ // The recovery moved the box and wrote the result this run answers
+ // with — an apply on top of that would overwrite that result and
+ // restart the edge a second time. The agent re-touches the marker
+ // if it still wants the update, and by then it can read the result.
+ return nil
+ }
req, err := a.readRequest()
if errors.Is(err, os.ErrNotExist) {
return nil
@@ -201,6 +216,15 @@ func (a *systemApplier) Run(ctx context.Context, recover bool) error {
a.ops.logf("system apply: request: %v", err)
return nil // nothing to record against: the request itself is unreadable
}
+ // A tuple this box has already rolled back is not tried again: the
+ // build failed its gate here, and re-applying it on every marker touch
+ // would restart the edge in a loop. Only those two outcomes stick —
+ // refused, deferred and noted are all worth another go.
+ if r, _ := readSystemResult(a.optDir); r.matches(a.channel, req.Version, req.PromotedAt) &&
+ (r.Outcome == systemOutcomeRolledBack || r.Outcome == systemOutcomeRollbackFailed) {
+ a.ops.logf("system apply: %s %s was already %s on this box; not applying it again", a.channel, req.Version, r.Outcome)
+ return nil
+ }
p, err := a.verifyPointer(ctx)
if err != nil {
return a.record(req, err)
@@ -370,7 +394,9 @@ func (a *systemApplier) prevBuild(ctx context.Context) string {
}
// stagePlan runs `<binary> bootstrap -root <dir> -plan-json` and returns
-// that binary's arming plan and the unit files it staged.
+// that binary's arming plan and the unit files it staged. The plan is left
+// in <dir>/plan.json next to them, so a later step can pick up the whole
+// staging (stagedPlan) without exec'ing the binary a second time.
func (a *systemApplier) stagePlan(ctx context.Context, binary, dir string) (bootstrapPlan, []string, error) {
os.RemoveAll(dir)
if err := os.MkdirAll(dir, 0o700); err != nil {
@@ -394,6 +420,9 @@ func (a *systemApplier) stagePlan(ctx context.Context, binary, dir string) (boot
units = append(units, e.Name())
}
}
+ if err := writeJSONDurable(filepath.Join(dir, "plan.json"), plan, 0o600); err != nil {
+ return bootstrapPlan{}, nil, err
+ }
return plan, units, nil
}
@@ -567,25 +596,29 @@ func (a *systemApplier) arm(plan bootstrapPlan) error {
// arm the previous plan. applied.json is untouched — the .prev build is
// what passed a gate, and it is what applied.json already names.
//
-// Every exit closes the journal: what a rollback could do it has done, and
-// a journal left open would have the next run (or the next boot) redo the
-// same restore against a box a human now has to look at anyway.
+// A rollback that cannot finish leaves the journal open: the box is then
+// between two builds — the previous binary under the failed build's units,
+// say — and only another recovery puts it right. The one exception is a
+// complete restore whose GATE failed: the files are all the previous
+// build's, restoring them again would change nothing, and the box needs a
+// human rather than a loop.
func (a *systemApplier) rollback(ctx context.Context, req systemRequest, reason string, j *systemJournal) error {
a.ops.logf("system apply: %s %s: rolling back: %s", a.channel, j.Target, reason)
failed := func(what string, err error) error {
_ = a.writeResult(req, systemOutcomeRollbackFailed, reason+"; "+what+": "+err.Error())
- a.removeJournal()
return fmt.Errorf("system apply: rollback of %s failed: %s: %v", j.Target, what, err)
}
plan, err := a.restoreFiles(ctx, j)
if err != nil {
- return failed("restore", err)
+ return failed("restore", err) // journal kept: the restore is half done
}
if err := a.restartEdge(); err != nil {
- return failed("restart", err)
+ return failed("restart", err) // journal kept: the edge is still to come back
}
if err := a.gate(ctx, j.PrevBuild); err != nil {
- return failed("previous build failed its gate too", err)
+ err := failed("previous build failed its gate too", err)
+ a.removeJournal()
+ return err
}
if err := a.arm(plan); err != nil {
a.ops.logf("system apply: arm previous plan: %v", err)
@@ -610,21 +643,48 @@ func (a *systemApplier) restoreFiles(ctx context.Context, j *systemJournal) (boo
}
// The same quiesce the install did: a root pass that is mid-flight
// against the failed build must not run on across the swap back.
- a.ops.systemctl(append([]string{"stop"}, systemQuiesceUnits...)...)
+ if _, err := a.ops.systemctl(append([]string{"stop"}, systemQuiesceUnits...)...); err != nil {
+ a.ops.logf("system apply: quiesce: %v (continuing)", err)
+ }
if err := copyFileDurable(prev, systemBinaryPath(a.optDir), 0o755); err != nil {
return bootstrapPlan{}, err
}
- // Staged from carlos.prev itself, not from the journal: the unit set to
- // put back is whatever the binary being restored ships.
+ // The unit set to put back is whatever the binary being restored ships,
+ // so it comes from carlos.prev and never from the journal. The install
+ // already staged it, and that staging is preferred: re-running a
+ // bootstrap on a box that is mid-rollback is one more thing that can
+ // fail, and it would wipe the good units before it tried.
prevDir := filepath.Join(a.stagingDir(), "prev")
- plan, _, err := a.stagePlan(ctx, prev, prevDir)
+ plan, err := stagedPlan(prevDir)
if err != nil {
- return bootstrapPlan{}, fmt.Errorf("carlos.prev bootstrap: %w", err)
+ if plan, _, err = a.stagePlan(ctx, prev, prevDir); err != nil {
+ return bootstrapPlan{}, fmt.Errorf("carlos.prev bootstrap: %w", err)
+ }
}
if err := a.installUnits(ctx, prevDir, j.UnitsAdded); err != nil {
return bootstrapPlan{}, err
}
- if _, err := a.ops.systemctl(append([]string{"enable"}, plan.Enabled...)...); err != nil {
+ if len(plan.Enabled) > 0 {
+ if _, err := a.ops.systemctl(append([]string{"enable"}, plan.Enabled...)...); err != nil {
+ return bootstrapPlan{}, err
+ }
+ }
+ return plan, nil
+}
+
+// stagedPlan reads back what a stagePlan into dir left there: the plan the
+// binary printed and the units it wrote. An empty unit directory or a
+// missing plan.json means the staging is not there to be reused.
+func stagedPlan(dir string) (bootstrapPlan, error) {
+ entries, err := os.ReadDir(filepath.Join(dir, "etc", "systemd", "system"))
+ if err != nil {
+ return bootstrapPlan{}, err
+ }
+ if len(entries) == 0 {
+ return bootstrapPlan{}, fmt.Errorf("no units staged in %s", dir)
+ }
+ var plan bootstrapPlan
+ if err := readJSONFile(filepath.Join(dir, "plan.json"), &plan); err != nil {
return bootstrapPlan{}, err
}
return plan, nil
@@ -667,36 +727,47 @@ func copyFileDurable(src, dst string, mode os.FileMode) error {
// that build proved itself, and undoing it would throw away a good apply.
// The journal is removed LAST, after the result is written and the restart
// issued, so a recovery that dies partway is recovered again.
-func (a *systemApplier) recoverIfNeeded(ctx context.Context) error {
+//
+// It reports whether it moved the box: a run that recovered has written the
+// result it answers with, and does not then go on to apply a request.
+func (a *systemApplier) recoverIfNeeded(ctx context.Context) (bool, error) {
j, err := readSystemJournal(a.optDir)
if err != nil || j == nil {
- return err
+ return false, err
}
- // The request is a hint here too: it may be gone, or name something
- // else entirely. The journal is what this recovery is about, so it
- // names the result when the request cannot.
+ // The result belongs to the apply being recovered, not to whatever the
+ // agent has asked for since. The request is used only when it is that
+ // same apply — then it carries the real promoted_at.
req, _ := a.readRequest()
- if req.Version == "" {
+ if req.Version != j.Target {
req = systemRequest{Version: j.Target, PromotedAt: j.StartedAt}
}
a.ops.logf("system apply: recovering an apply of %s interrupted at %s", j.Target, j.Phase)
- switch j.Phase {
- case systemPhaseGated:
- plan, _, err := a.stagePlan(ctx, systemBinaryPath(a.optDir), filepath.Join(a.stagingDir(), "new"))
- if err != nil {
- return fmt.Errorf("system apply: recovery: %w", err)
- }
- return a.finish(req, j.Target, plan)
- case systemPhaseBackingUp:
+ if j.Phase == systemPhaseBackingUp {
// Nothing on the box had changed yet: only a half-written .prev
- // copy can be left, and it is not the file anything reads.
+ // copy can be left, and it is not the file anything reads. The run
+ // carries on to the request it was started for.
os.Remove(systemPrevPath(a.optDir) + ".tmp")
a.removeJournal()
- return nil
+ return false, nil
+ }
+ // Forward from gated — but only while the build that passed the gate is
+ // still the one on disk. A binary that has moved since means this is a
+ // pre-gate state whatever the journal says.
+ if j.Phase == systemPhaseGated {
+ if got := a.onDiskBuild(ctx); got != j.Target {
+ a.ops.logf("system apply: recovery: %s passed its gate but the binary on disk reports %q; rolling back instead", j.Target, got)
+ } else {
+ plan, _, err := a.stagePlan(ctx, systemBinaryPath(a.optDir), filepath.Join(a.stagingDir(), "new"))
+ if err != nil {
+ return false, fmt.Errorf("system apply: recovery: %w", err)
+ }
+ return true, a.finish(req, j.Target, plan)
+ }
}
plan, err := a.restoreFiles(ctx, j)
if err != nil {
- return fmt.Errorf("system apply: recovery: %w", err) // journal kept: the next run recovers again
+ return false, fmt.Errorf("system apply: recovery: %w", err) // journal kept: the next run recovers again
}
if a.ops.edgeStartedThisBoot() {
a.ops.systemctl("reset-failed", systemEdgeUnit)
@@ -704,7 +775,7 @@ func (a *systemApplier) recoverIfNeeded(ctx context.Context) error {
// handler, and a blocking restart of the unit that triggered it
// would wait on the job that is waiting on this one.
if _, err := a.ops.systemctl("restart", "--no-block", systemEdgeUnit); err != nil {
- return fmt.Errorf("system apply: recovery: restart: %w", err)
+ return false, fmt.Errorf("system apply: recovery: restart: %w", err)
}
var armed []string
for _, u := range plan.Enabled {
@@ -717,10 +788,10 @@ func (a *systemApplier) recoverIfNeeded(ctx context.Context) error {
}
}
if err := a.writeResult(req, systemOutcomeRolledBack, "interrupted at "+j.Phase); err != nil {
- return err
+ return false, err
}
a.removeJournal()
- return nil
+ return true, nil
}
// record turns a refusal or deferral into result.json and a nil return.
diff --git a/internal/host/systemapply_test.go b/internal/host/systemapply_test.go
index e50b75ce5..c27f30492 100644
--- a/internal/host/systemapply_test.go
+++ b/internal/host/systemapply_test.go
@@ -671,6 +671,11 @@ func TestApplyRollsBackWhenTheGateFails(t *testing.T) {
if !strings.Contains(joined, "systemctl disable --now carlos-new-only.path") {
t.Error("added unit not disabled")
}
+ // The install already staged carlos.prev; the rollback reuses it rather
+ // than re-running a bootstrap on a box that is already unhealthy.
+ if n := strings.Count(joined, "run carlos.prev bootstrap"); n != 1 {
+ t.Errorf("carlos.prev bootstrapped %d times, want 1:\n%s", n, joined)
+ }
// reset-failed precedes the rollback's restart too, and the previous
// plan is armed afterwards.
i := strings.LastIndex(joined, "systemctl reset-failed carlos-edge")
@@ -701,8 +706,11 @@ func TestApplyRollbackFailedLeavesPrevAndAppliedAlone(t *testing.T) {
t.Fatal(".prev damaged")
}
// A second apply of a NEW version must not refresh .prev from the
- // unproven disk binary.
- f.answer["version:"+systemBinaryPath(f.a.optDir)] = "carlos v1\n" // rollback restored v1 on disk
+ // unproven disk binary. The box a failed rollback leaves behind is the
+ // one that makes that matter: applied.json still names v1, and the v2
+ // that never passed a gate is what is on disk.
+ os.WriteFile(systemBinaryPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v2\n"), 0o755)
+ f.answer["version:"+systemBinaryPath(f.a.optDir)] = "carlos v2\n"
os.Remove(systemRequestPath(f.a.dataDir))
f.promote("v3", []byte("#!/bin/sh\necho carlos v3\n"), false)
f.answer["version:"+filepath.Join(f.a.stagingDir(), "carlos")] = "carlos v3\n"
@@ -795,7 +803,39 @@ func TestRecoveryFinishesForwardFromGated(t *testing.T) {
}
func TestRecoveryRemovesTheJournalLast(t *testing.T) {
- // A restart that fails mid-recovery leaves the journal for the next run.
+ // The restore is done and the result is the last thing before the
+ // journal goes: a restart that fails after the files are back leaves
+ // both the journal and no result at all for the next run to redo.
+ f, _ := applyHappy(t)
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v1\n"), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
+ systemJournal{Phase: systemPhaseInstalled, Target: "v2", PrevBuild: "v1", StartedAt: f.now}, 0o600)
+ f.edgeStartedThisBoot = true
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "restart" {
+ return "", errors.New("job for carlos-edge.service failed")
+ }
+ return orig(args...)
+ }
+ if err := f.a.Run(context.Background(), true); err == nil {
+ t.Fatal("a failed recovery must exit non-zero")
+ }
+ if j, _ := readSystemJournal(f.a.optDir); j == nil {
+ t.Fatal("journal removed although recovery did not complete")
+ }
+ if r := f.result(); r != nil {
+ t.Fatalf("result written before the restart was issued: %+v", r)
+ }
+ bin, _ := os.ReadFile(systemBinaryPath(f.a.optDir))
+ if !strings.Contains(string(bin), "carlos v1") {
+ t.Fatal("the restore itself did not happen")
+ }
+}
+
+func TestRecoveryKeepsTheJournalWhenTheRestoreFails(t *testing.T) {
+ // systemd will not reload, so the units are not back: the journal stays
+ // open and the next run recovers again.
f, _ := applyHappy(t)
os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v1\n"), 0o755)
writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
@@ -814,3 +854,141 @@ func TestRecoveryRemovesTheJournalLast(t *testing.T) {
t.Fatal("journal removed although recovery did not complete")
}
}
+
+func TestRecoveryAnswersTheJournalsVersionNotALaterRequest(t *testing.T) {
+ // The agent has moved on to v3 while a v2 apply lay interrupted. The
+ // result belongs to v2 — v3 has not been tried at all.
+ f, _ := applyHappy(t)
+ started := f.now
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v1\n"), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
+ systemJournal{Phase: systemPhaseUnits, Target: "v2", PrevBuild: "v1", StartedAt: started}, 0o600)
+ v3 := f.promote("v3", []byte("#!/bin/sh\necho carlos v3\n"), false)
+ if err := f.a.Run(context.Background(), true); err != nil {
+ t.Fatal(err)
+ }
+ r := f.result()
+ if r == nil || r.Version != "v2" || !r.PromotedAt.Equal(started) {
+ t.Fatalf("result = %+v, want the v2 the journal names at %s", r, started)
+ }
+ if r.PromotedAt.Equal(v3.PromotedAt) {
+ t.Fatal("the recovery answered for the request the agent wrote later")
+ }
+}
+
+func TestRecoveryFromGatedRollsBackWhenTheBinaryIsNotTheGatedOne(t *testing.T) {
+ // The journal says v2 passed its gate, but v1 is on disk: whatever put
+ // it there, this is a pre-gate state and going forward would record an
+ // apply of a build that is not installed.
+ f, _ := applyHappy(t)
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v1\n"), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
+ systemJournal{Phase: systemPhaseGated, Target: "v2", PrevBuild: "v1", StartedAt: f.now}, 0o600)
+ if err := f.a.Run(context.Background(), true); err != nil {
+ t.Fatal(err)
+ }
+ if ap, _ := readSystemApplied(f.a.optDir); ap != nil {
+ t.Fatalf("applied.json written for a build that is not on disk: %+v", ap)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRolledBack {
+ t.Fatalf("result = %+v", r)
+ }
+}
+
+func TestRollbackThatCannotRestoreKeepsTheJournalForTheNextRun(t *testing.T) {
+ f, _ := applyHappy(t)
+ f.edgeUp = false
+ prevDir := filepath.Join(f.a.stagingDir(), "prev")
+ prevPlan := f.answer["plan:"+systemPrevPath(f.a.optDir)]
+ stops := 0
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ // The rollback's own quiesce: by then the staging is gone and
+ // carlos.prev will not bootstrap, so the restore stops after the
+ // binary and before the units.
+ if args[0] == "stop" {
+ if stops++; stops == 2 {
+ os.RemoveAll(prevDir)
+ delete(f.answer, "plan:"+systemPrevPath(f.a.optDir))
+ }
+ }
+ return orig(args...)
+ }
+ if err := f.a.Run(context.Background(), false); err == nil {
+ t.Fatal("a failed rollback must exit non-zero")
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRollbackFailed {
+ t.Fatalf("result = %+v", r)
+ }
+ if j, _ := readSystemJournal(f.a.optDir); j == nil {
+ t.Fatal("journal removed although the restore never finished")
+ }
+ units := filepath.Join(f.a.etcDir, "systemd", "system")
+ if _, err := os.Stat(filepath.Join(units, "carlos-new-only.path")); err != nil {
+ t.Fatal("precondition: the failed build's units are what is still installed")
+ }
+ // The next run recovers what the rollback could not: the bootstrap
+ // works again, and the previous build's units go back.
+ f.answer["plan:"+systemPrevPath(f.a.optDir)] = prevPlan
+ if err := f.a.Run(context.Background(), true); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Stat(filepath.Join(units, "carlos-new-only.path")); err == nil {
+ t.Error("added unit survived the recovery")
+ }
+ if _, err := os.Stat(filepath.Join(units, "carlos-old-only.timer")); err != nil {
+ t.Error("removed unit not restored by the recovery")
+ }
+ if j, _ := readSystemJournal(f.a.optDir); j != nil {
+ t.Error("journal left open after a complete recovery")
+ }
+}
+
+func TestARecoveredRunDoesNotThenApplyTheRequest(t *testing.T) {
+ // The v2 request is still on disk, and the recovery has just rolled v2
+ // back. Applying it again in the same run would overwrite the result
+ // the agent is waiting for and restart the edge a second time.
+ f, _ := applyHappy(t)
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v1\n"), 0o755)
+ os.WriteFile(systemBinaryPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v2\n"), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
+ systemJournal{Phase: systemPhaseUnits, Target: "v2", PrevBuild: "v1", StartedAt: f.now}, 0o600)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRolledBack {
+ t.Fatalf("result = %+v", r)
+ }
+ if ap, _ := readSystemApplied(f.a.optDir); ap != nil {
+ t.Fatalf("the request was applied on top of the recovery: %+v", ap)
+ }
+ bin, _ := os.ReadFile(systemBinaryPath(f.a.optDir))
+ if !strings.Contains(string(bin), "carlos v1") {
+ t.Fatal("the binary is not the one the recovery restored")
+ }
+}
+
+func TestARolledBackTupleIsNotAppliedAgain(t *testing.T) {
+ f, req := applyHappy(t)
+ results := filepath.Join(systemRootDir(f.a.optDir), "result.json")
+ writeJSONDurable(results, systemResult{Channel: "stable", Version: req.Version, PromotedAt: req.PromotedAt,
+ Outcome: systemOutcomeRolledBack, Reason: "gate: carlos-edge is failed", At: f.now}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if len(f.calls) != 0 {
+ t.Fatalf("a build that already rolled back here was applied again: %v", f.calls)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRolledBack {
+ t.Fatalf("result = %+v, the standing outcome must be left alone", r)
+ }
+ // A refusal is not terminal: the same tuple is tried again.
+ writeJSONDurable(results, systemResult{Channel: "stable", Version: req.Version, PromotedAt: req.PromotedAt,
+ Outcome: systemOutcomeRefused, Reason: "seen.json unreadable", At: f.now}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeApplied {
+ t.Fatalf("result = %+v, a refused tuple must stay retryable", r)
+ }
+}