amadan

carlos / platform Public

host: carlos system apply — rollback, recovery, and the real box

Rollback restores carlos.prev and its unit set, gates it, arms the
previous plan, and leaves applied.json alone. Recovery is files-only
and goes forward from phase gated; it restarts the edge only when the
edge has started this boot, and removes the journal last.

The production probe is plain HTTP on 127.0.0.1:80 — the edge's :80
listener (ACME HTTP-01 plus the redirect to https) runs on every box,
so an answer there proves the process serves.
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 6017307a49d0ef98556c44b7bf596855eee48707 parent 904e016
2 files changed, +396 −8
  • internal/host/systemapply.go +205 −8
  • internal/host/systemapply_test.go +191 −0
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index 6440bc2f0..39def0f57 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -7,14 +7,21 @@ import (
"encoding/hex"
"encoding/json"
"errors"
+ "flag"
"fmt"
"io"
+ "log"
+ "net/http"
"os"
+ "os/exec"
"path/filepath"
+ "runtime"
"strings"
"time"
+ "github.com/carlosframework/platform/internal/cliflag"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/version"
)
// The root half of the platform self-update (spec §6). Runs as root from
@@ -63,6 +70,78 @@ type sysOps struct {
logf func(format string, args ...any)
}
+// RunSystemApply is `carlos system apply [-recover]`: the root oneshot
+// behind carlos-system-update.service, and with -recover the boot and
+// OnFailure= unit that puts an interrupted apply back together.
+func RunSystemApply(args []string) {
+ fs := flag.NewFlagSet("system apply", flag.ExitOnError)
+ recover := fs.Bool("recover", false, "recover an interrupted apply (files only) and exit: the boot and OnFailure unit")
+ fs.Usage = cliflag.Usage(fs, "")
+ _ = fs.Parse(args)
+ if os.Geteuid() != 0 {
+ log.Fatal("carlos system apply is the root oneshot behind carlos-system-update.service; it is not for operators")
+ }
+ ctx := context.Background()
+ store, err := openDeployStore(ctx)
+ if err != nil {
+ log.Fatalf("system apply: %v", err)
+ }
+ pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
+ if err != nil {
+ log.Fatalf("system apply: %v", err)
+ }
+ a := &systemApplier{
+ ops: productionSysOps(), store: store, pub: pub, channel: systemChannel(os.Getenv),
+ optDir: release.OptDir(), dataDir: dataDir(), etcDir: "/etc", build: version.Build,
+ platform: runtime.GOOS + "-" + runtime.GOARCH, gateWait: systemGateWait, soak: systemGateSoak,
+ }
+ if err := a.Run(ctx, *recover); err != nil {
+ log.Fatalf("%v", err)
+ }
+}
+
+// productionSysOps is the real box behind the seam.
+func productionSysOps() sysOps {
+ return sysOps{
+ systemctl: func(args ...string) (string, error) {
+ out, err := exec.Command("systemctl", args...).CombinedOutput()
+ if err != nil {
+ return string(out), fmt.Errorf("systemctl %s: %w: %s", strings.Join(args, " "), err, strings.TrimSpace(string(out)))
+ }
+ return string(out), nil
+ },
+ run: func(ctx context.Context, timeout time.Duration, name string, args ...string) (string, error) {
+ ctx, cancel := context.WithTimeout(ctx, timeout)
+ defer cancel()
+ out, err := exec.CommandContext(ctx, name, args...).Output()
+ if err != nil {
+ return string(out), fmt.Errorf("%s %s: %w", name, strings.Join(args, " "), err)
+ }
+ return string(out), nil
+ },
+ now: time.Now,
+ sleep: time.Sleep,
+ probeEdge: func() error {
+ // The edge's :80 listener answers on every box — ACME HTTP-01
+ // and the redirect to https — so any HTTP answer on loopback
+ // proves the process serves. A redirect is an answer, and a box
+ // with no route yet must still pass (spec §6 step 10).
+ c := &http.Client{Timeout: 3 * time.Second, CheckRedirect: func(*http.Request, []*http.Request) error { return http.ErrUseLastResponse }}
+ resp, err := c.Get("http://127.0.0.1/")
+ if err != nil {
+ return err
+ }
+ resp.Body.Close()
+ return nil
+ },
+ edgeStartedThisBoot: func() bool {
+ out, err := exec.Command("systemctl", "show", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit).Output()
+ return err == nil && strings.TrimSpace(string(out)) != "" && strings.TrimSpace(string(out)) != "0"
+ },
+ logf: log.Printf,
+ }
+}
+
type systemApplier struct {
ops sysOps
store release.Store
@@ -169,7 +248,9 @@ func (a *systemApplier) install(ctx context.Context, req systemRequest, p *relea
if err != nil {
return a.record(req, fmt.Errorf("%w: candidate bootstrap: %v", errSystemRefused, err))
}
- prevPlan, prevUnits, err := a.stagePlan(ctx, systemPrevPath(a.optDir), prevDir)
+ // Only the unit set is kept: a rollback re-runs this bootstrap, so that
+ // what it puts back is derived from the binary it is putting back.
+ _, prevUnits, err := a.stagePlan(ctx, systemPrevPath(a.optDir), prevDir)
if err != nil {
return a.record(req, fmt.Errorf("%w: carlos.prev bootstrap: %v", errSystemRefused, err))
}
@@ -179,7 +260,7 @@ func (a *systemApplier) install(ctx context.Context, req systemRequest, p *relea
return err
}
fail := func(step string, err error) error {
- return a.rollback(ctx, req, fmt.Sprintf("%s: %v", step, err), &j, prevPlan)
+ return a.rollback(ctx, req, fmt.Sprintf("%s: %v", step, err), &j)
}
// Stop the root passes that exec the old binary against the old edge
// before the rename; one that is already running is killed, and those
@@ -482,11 +563,71 @@ func (a *systemApplier) arm(plan bootstrapPlan) error {
return nil
}
-// rollback is step 13 (Task 8c). Until then it records the outcome the
-// box is actually in and fails loudly rather than pretending.
-func (a *systemApplier) rollback(ctx context.Context, req systemRequest, reason string, j *systemJournal, prevPlan bootstrapPlan) error {
- _ = a.writeResult(req, systemOutcomeRolledBack, reason)
- return errors.New("rollback not implemented: " + reason)
+// rollback is step 13: back to carlos.prev and its units, restart, gate,
+// 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.
+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)
+ }
+ if err := a.restartEdge(); err != nil {
+ return failed("restart", err)
+ }
+ if err := a.gate(ctx, j.PrevBuild); err != nil {
+ return failed("previous build failed its gate too", err)
+ }
+ if err := a.arm(plan); err != nil {
+ a.ops.logf("system apply: arm previous plan: %v", err)
+ }
+ if err := a.writeResult(req, systemOutcomeRolledBack, reason); err != nil {
+ return err
+ }
+ a.removeJournal()
+ return fmt.Errorf("system apply: %s rolled back: %s", j.Target, reason)
+}
+
+// restoreFiles is the files-only half shared by rollback and recovery:
+// carlos.prev over carlos (keeping .prev), units_added disabled and
+// removed, the previous unit set copied back (bringing units_removed back),
+// sysusers, tmpfiles, daemon-reload, and `enable` of the previous plan so
+// it comes up at the next boot. No restart, no start: what to do about the
+// running edge is the caller's, and at boot the answer is nothing.
+func (a *systemApplier) restoreFiles(ctx context.Context, j *systemJournal) (bootstrapPlan, error) {
+ prev := systemPrevPath(a.optDir)
+ if _, err := os.Stat(prev); err != nil {
+ return bootstrapPlan{}, fmt.Errorf("no carlos.prev: %w", err)
+ }
+ // 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 := 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.
+ prevDir := filepath.Join(a.stagingDir(), "prev")
+ plan, _, err := a.stagePlan(ctx, prev, prevDir)
+ if 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 {
+ return bootstrapPlan{}, err
+ }
+ return plan, nil
}
// unitsMinus is have − drop, order preserved.
@@ -518,12 +659,68 @@ func copyFileDurable(src, dst string, mode os.FileMode) error {
})
}
+// recoverIfNeeded is step 0: an open journal means a previous run was cut
+// off, and nothing else happens until the box is whole again. Before the
+// gate → back, files only, restarting the edge only when the edge has
+// already started this boot (at boot, systemd is about to start it and a
+// restart from an ordering-before unit would deadlock). At gated → forward:
+// 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 {
j, err := readSystemJournal(a.optDir)
if err != nil || j == nil {
return err
}
- return fmt.Errorf("system apply: journal open at %s — recovery not implemented", j.Phase) // Task 8c
+ // 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.
+ req, _ := a.readRequest()
+ if req.Version == "" {
+ 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:
+ // Nothing on the box had changed yet: only a half-written .prev
+ // copy can be left, and it is not the file anything reads.
+ os.Remove(systemPrevPath(a.optDir) + ".tmp")
+ a.removeJournal()
+ return nil
+ }
+ plan, err := a.restoreFiles(ctx, j)
+ if err != nil {
+ return fmt.Errorf("system apply: recovery: %w", err) // journal kept: the next run recovers again
+ }
+ if a.ops.edgeStartedThisBoot() {
+ a.ops.systemctl("reset-failed", systemEdgeUnit)
+ // --no-block: this can be running as the edge's own OnFailure=
+ // 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)
+ }
+ var armed []string
+ for _, u := range plan.Enabled {
+ if strings.HasSuffix(u, ".timer") || strings.HasSuffix(u, ".path") {
+ armed = append(armed, u)
+ }
+ }
+ if len(armed) > 0 {
+ a.ops.systemctl(append([]string{"start"}, armed...)...)
+ }
+ }
+ if err := a.writeResult(req, systemOutcomeRolledBack, "interrupted at "+j.Phase); err != nil {
+ return err
+ }
+ a.removeJournal()
+ return 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 d3b8cd4f0..e50b75ce5 100644
--- a/internal/host/systemapply_test.go
+++ b/internal/host/systemapply_test.go
@@ -623,3 +623,194 @@ func TestApplyRecordsAnArmingFailureAgainstTheApply(t *testing.T) {
t.Fatalf("applied = %+v (the gate passed; the roll stands)", ap)
}
}
+
+func TestApplyRollsBackWhenTheGateFails(t *testing.T) {
+ f, req := applyHappy(t)
+ // v2 never comes up; v1 does — which is what makes this a rollback and
+ // not a rollback-failed. The fake edge answers for whatever binary is
+ // on disk when the restart is issued.
+ f.edgeUp = false
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "restart" {
+ if b, _ := os.ReadFile(systemBinaryPath(f.a.optDir)); strings.Contains(string(b), "carlos v1") {
+ f.edgeUp = true
+ }
+ }
+ return orig(args...)
+ }
+ err := f.a.Run(context.Background(), false)
+ if err == nil {
+ t.Fatal("a failed gate must exit non-zero")
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeRolledBack || !r.matches("stable", "v2", req.PromotedAt) || !strings.Contains(r.Reason, "gate") {
+ t.Fatalf("result = %+v", r)
+ }
+ bin, _ := os.ReadFile(systemBinaryPath(f.a.optDir))
+ if !strings.Contains(string(bin), "carlos v1") {
+ t.Fatal("binary not restored")
+ }
+ if _, err := os.Stat(systemPrevPath(f.a.optDir)); err != nil {
+ t.Fatal(".prev consumed by the rollback")
+ }
+ units := filepath.Join(f.a.etcDir, "systemd", "system")
+ if _, err := os.Stat(filepath.Join(units, "carlos-new-only.path")); err == nil {
+ t.Error("added unit survived the rollback")
+ }
+ if _, err := os.Stat(filepath.Join(units, "carlos-old-only.timer")); err != nil {
+ t.Error("removed unit not restored")
+ }
+ if ap, _ := readSystemApplied(f.a.optDir); ap != nil {
+ t.Fatalf("applied.json written by a rollback: %+v", ap)
+ }
+ if j, _ := readSystemJournal(f.a.optDir); j != nil {
+ t.Fatal("journal left open after rollback")
+ }
+ joined := strings.Join(f.calls, "\n")
+ if !strings.Contains(joined, "systemctl disable --now carlos-new-only.path") {
+ t.Error("added unit not disabled")
+ }
+ // reset-failed precedes the rollback's restart too, and the previous
+ // plan is armed afterwards.
+ i := strings.LastIndex(joined, "systemctl reset-failed carlos-edge")
+ k := strings.LastIndex(joined, "systemctl restart carlos-edge")
+ m := strings.LastIndex(joined, "systemctl start carlos-adopt.timer")
+ if !(i < k && k < m) {
+ t.Fatalf("rollback order wrong:\n%s", joined)
+ }
+}
+
+func TestApplyRollbackFailedLeavesPrevAndAppliedAlone(t *testing.T) {
+ f, req := applyHappy(t)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "applied.json"), systemApplied{Version: "v1", Channel: "stable", At: f.now}, 0o644)
+ f.edgeUp = false // never comes up, on either binary
+ if err := f.a.Run(context.Background(), false); err == nil {
+ t.Fatal("must exit non-zero")
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeRollbackFailed || !r.matches("stable", "v2", req.PromotedAt) {
+ t.Fatalf("result = %+v", r)
+ }
+ ap, _ := readSystemApplied(f.a.optDir)
+ if ap == nil || ap.Version != "v1" {
+ t.Fatalf("applied = %+v, must still name v1", ap)
+ }
+ prev, _ := os.ReadFile(systemPrevPath(f.a.optDir))
+ if !strings.Contains(string(prev), "carlos v1") {
+ 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
+ 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"
+ f.answer["plan:"+filepath.Join(f.a.stagingDir(), "carlos")] = f.answer["plan:"+systemPrevPath(f.a.optDir)]
+ f.answer["units:"+filepath.Join(f.a.stagingDir(), "carlos")] = f.answer["units:"+systemPrevPath(f.a.optDir)]
+ _ = f.a.Run(context.Background(), false)
+ prev2, _ := os.ReadFile(systemPrevPath(f.a.optDir))
+ if string(prev2) != string(prev) {
+ t.Fatal(".prev refreshed after rollback-failed")
+ }
+}
+
+func TestRecoveryRollsBackFilesOnlyAtBootAndRestartsAtRuntime(t *testing.T) {
+ for _, phase := range []string{systemPhaseInstalling, systemPhaseInstalled, systemPhaseUnits, systemPhaseRestarted} {
+ t.Run(phase, func(t *testing.T) {
+ f, _ := applyHappy(t)
+ // Simulate: v2 installed, journal open at phase, box rebooted.
+ 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)
+ os.WriteFile(filepath.Join(f.a.etcDir, "systemd", "system", "carlos-new-only.path"), []byte("x"), 0o644)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "journal.json"),
+ systemJournal{Phase: phase, Target: "v2", PrevBuild: "v1", UnitsAdded: []string{"carlos-new-only.path"}, UnitsRemoved: []string{"carlos-old-only.timer"}, StartedAt: f.now}, 0o600)
+ f.edgeStartedThisBoot = false
+ if err := f.a.Run(context.Background(), true); err != nil {
+ t.Fatal(err)
+ }
+ bin, _ := os.ReadFile(systemBinaryPath(f.a.optDir))
+ if !strings.Contains(string(bin), "carlos v1") {
+ t.Fatal("binary not restored")
+ }
+ if _, err := os.Stat(filepath.Join(f.a.etcDir, "systemd", "system", "carlos-new-only.path")); err == nil {
+ t.Error("added unit survived")
+ }
+ if _, err := os.Stat(filepath.Join(f.a.etcDir, "systemd", "system", "carlos-old-only.timer")); err != nil {
+ t.Error("removed unit not restored")
+ }
+ joined := strings.Join(f.calls, "\n")
+ if strings.Contains(joined, "systemctl restart carlos-edge") || strings.Contains(joined, "systemctl start ") {
+ t.Fatalf("boot recovery restarted or started something:\n%s", joined)
+ }
+ if !strings.Contains(joined, "systemctl enable carlos-edge.service") {
+ t.Fatal("previous plan not enabled for the boot")
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeRolledBack || !strings.Contains(r.Reason, phase) {
+ t.Fatalf("result = %+v", r)
+ }
+ if j, _ := readSystemJournal(f.a.optDir); j != nil {
+ t.Fatal("journal left open")
+ }
+ })
+ }
+ // Runtime (OnFailure=): same, plus a non-blocking restart and the
+ // previous plan's timers/paths started.
+ 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: systemPhaseUnits, Target: "v2", PrevBuild: "v1", StartedAt: f.now}, 0o600)
+ f.edgeStartedThisBoot = true
+ if err := f.a.Run(context.Background(), true); err != nil {
+ t.Fatal(err)
+ }
+ joined := strings.Join(f.calls, "\n")
+ for _, want := range []string{"systemctl stop carlos-adopt.timer", "systemctl reset-failed carlos-edge", "systemctl restart --no-block carlos-edge", "systemctl start carlos-adopt.timer"} {
+ if !strings.Contains(joined, want) {
+ t.Errorf("runtime recovery lacks %q:\n%s", want, joined)
+ }
+ }
+}
+
+func TestRecoveryFinishesForwardFromGated(t *testing.T) {
+ f, req := applyHappy(t)
+ os.WriteFile(systemBinaryPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v2\n"), 0o755)
+ writeJSONDurable(systemRequestPath(f.a.dataDir), req, 0o644)
+ 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)
+ }
+ bin, _ := os.ReadFile(systemBinaryPath(f.a.optDir))
+ if !strings.Contains(string(bin), "carlos v2") {
+ t.Fatal("a gated apply was undone")
+ }
+ ap, _ := readSystemApplied(f.a.optDir)
+ if ap == nil || ap.Version != "v2" {
+ t.Fatalf("applied = %+v", ap)
+ }
+ if r := f.result(); r == nil || r.Outcome != systemOutcomeApplied {
+ t.Fatalf("result = %+v", r)
+ }
+}
+
+func TestRecoveryRemovesTheJournalLast(t *testing.T) {
+ // A restart that fails mid-recovery leaves the journal for the next run.
+ 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)
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "daemon-reload" {
+ return "", errors.New("systemd busy")
+ }
+ 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")
+ }
+}