amadan

carlos / platform Public

host: the soak demands zero restarts; arming failures are recorded; one durable-write protocol

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 904e0169012b087f0448fc55d77149524693f4a8 parent 6758dde
3 files changed, +184 −62
  • internal/host/systemapply.go +77 −55
  • internal/host/systemapply_test.go +83 −1
  • internal/host/systemstate.go +24 −6
diff --git a/internal/host/systemapply.go b/internal/host/systemapply.go
index 1513756a9..6440bc2f0 100644
--- a/internal/host/systemapply.go
+++ b/internal/host/systemapply.go
@@ -174,7 +174,7 @@ func (a *systemApplier) install(ctx context.Context, req systemRequest, p *relea
return a.record(req, fmt.Errorf("%w: carlos.prev bootstrap: %v", errSystemRefused, err))
}
j := systemJournal{Phase: systemPhaseInstalling, Target: target, PrevBuild: a.prevBuild(ctx),
- UnitsAdded: diff(newUnits, prevUnits), UnitsRemoved: diff(prevUnits, newUnits), StartedAt: a.ops.now().UTC()}
+ UnitsAdded: unitsMinus(newUnits, prevUnits), UnitsRemoved: unitsMinus(prevUnits, newUnits), StartedAt: a.ops.now().UTC()}
if err := a.journal(j); err != nil {
return err
}
@@ -197,7 +197,7 @@ func (a *systemApplier) install(ctx context.Context, req systemRequest, p *relea
if err := a.journal(j); err != nil {
return fail("journal", err)
}
- if err := a.installUnits(newDir, j.UnitsRemoved); err != nil {
+ if err := a.installUnits(ctx, newDir, j.UnitsRemoved); err != nil {
return fail("units", err)
}
j.Phase = systemPhaseUnits
@@ -223,17 +223,34 @@ func (a *systemApplier) install(ctx context.Context, req systemRequest, p *relea
// finish is steps 11–12, and what recovery runs forward from phase gated.
func (a *systemApplier) finish(req systemRequest, target string, plan bootstrapPlan) error {
+ reason := ""
if err := a.arm(plan); err != nil {
+ // The gate passed, so this run is an apply and rolling back now
+ // would undo a build that proved itself. But `enable` is one
+ // batched call: one bad unit leaves the box on the new binary with
+ // nothing armed, and a fleet-visible reason is the only way anyone
+ // finds out.
a.ops.logf("system apply: arm: %v", err)
+ reason = "armed with errors: " + err.Error()
}
if err := writeJSONDurable(a.file("applied.json"), systemApplied{Version: target, Channel: a.channel, At: a.ops.now().UTC()}, 0o644); err != nil {
return err
}
- if err := a.writeResult(req, systemOutcomeApplied, ""); err != nil {
+ if err := a.writeResult(req, systemOutcomeApplied, reason); err != nil {
return err
}
+ a.removeJournal()
+ return nil
+}
+
+// removeJournal closes the transaction, durably: an unsynced removal can
+// come back after a power loss, and a journal that outlives the run it
+// belongs to is one the next run recovers from.
+func (a *systemApplier) removeJournal() {
os.Remove(a.file("journal.json"))
- return syncDirBestEffort(a.rootDir())
+ if err := syncDir(a.rootDir()); err != nil {
+ a.ops.logf("system apply: fsync %s: %v", a.rootDir(), err)
+ }
}
// decidePrev refreshes carlos.prev from the on-disk binary only when that
@@ -259,7 +276,7 @@ func (a *systemApplier) decidePrev(ctx context.Context) (bool, error) {
if err := copyFileDurable(systemBinaryPath(a.optDir), prev, 0o755); err != nil {
return false, err
}
- os.Remove(a.file("journal.json"))
+ a.removeJournal()
return true, nil
}
@@ -308,8 +325,11 @@ func (a *systemApplier) journal(j systemJournal) error {
// group does not resolve, and carlos-edge never starts again after the
// next boot — then daemon-reload. Never the staging root's data/: it holds
// a 0-byte registry.db.
-func (a *systemApplier) installUnits(stagedRoot string, remove []string) error {
+func (a *systemApplier) installUnits(ctx context.Context, stagedRoot string, remove []string) error {
unitDir := filepath.Join(a.etcDir, "systemd", "system")
+ if err := os.MkdirAll(unitDir, 0o755); err != nil {
+ return err
+ }
for _, u := range remove {
if _, err := a.ops.systemctl("disable", "--now", u); err != nil {
a.ops.logf("system apply: disable %s: %v", u, err)
@@ -337,7 +357,7 @@ func (a *systemApplier) installUnits(stagedRoot string, remove []string) error {
if err := copyFileDurable(sys, dst, 0o644); err != nil {
return err
}
- if _, err := a.ops.run(context.Background(), systemVersionWait, "systemd-sysusers", dst); err != nil {
+ if _, err := a.ops.run(ctx, systemVersionWait, "systemd-sysusers", dst); err != nil {
return fmt.Errorf("systemd-sysusers: %w", err)
}
}
@@ -349,7 +369,7 @@ func (a *systemApplier) installUnits(stagedRoot string, remove []string) error {
if err := copyFileDurable(tmp, dst, 0o644); err != nil {
return err
}
- if _, err := a.ops.run(context.Background(), systemVersionWait, "systemd-tmpfiles", "--create", dst); err != nil {
+ if _, err := a.ops.run(ctx, systemVersionWait, "systemd-tmpfiles", "--create", dst); err != nil {
a.ops.logf("system apply: systemd-tmpfiles: %v (continuing, as the script does)", err)
}
_, err = a.ops.systemctl("daemon-reload")
@@ -365,9 +385,11 @@ func (a *systemApplier) restartEdge() error {
}
// gate: within gateWait the edge is active, answers on loopback, and the
-// installed binary reports the target; then a soak across which neither
-// NRestarts nor ActiveEnterTimestampMonotonic moves, so a build that
-// starts, answers and dies a second later is caught here.
+// installed binary reports the target; then a soak across which NRestarts
+// stays 0 — step 9's reset-failed zeroed it, so anything above 0 is a
+// crash the new build caused — and ActiveEnterTimestampMonotonic does not
+// move. A build that starts, answers, and dies a second later is caught
+// here, and so is one that crashed once and then settled.
func (a *systemApplier) gate(ctx context.Context, target string) error {
wait, soak := a.gateWait, a.soak
if wait <= 0 {
@@ -388,24 +410,56 @@ func (a *systemApplier) gate(ctx context.Context, target string) error {
}
a.ops.sleep(time.Second)
}
- before, err := a.ops.systemctl("show", "-p", "NRestarts", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit)
+ before, err := a.edgeRunState()
if err != nil {
return err
}
+ if err := before.steady(); err != nil {
+ return err
+ }
a.ops.sleep(soak)
- after, err := a.ops.systemctl("show", "-p", "NRestarts", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit)
+ after, err := a.edgeRunState()
if err != nil {
return err
}
- if strings.TrimSpace(before) != strings.TrimSpace(after) {
- return fmt.Errorf("edge restarted during the soak (%s → %s)",
- strings.ReplaceAll(strings.TrimSpace(before), "\n", "/"), strings.ReplaceAll(strings.TrimSpace(after), "\n", "/"))
+ if err := after.steady(); err != nil {
+ return err
+ }
+ if before != after {
+ return fmt.Errorf("edge restarted during the soak (%s → %s)", before, after)
}
return a.gateOnce(ctx, target)
}
+// edgeRunState is the edge's NRestarts and ActiveEnterTimestampMonotonic,
+// in that order — the order `show -p` is asked for them.
+type edgeRunState struct{ restarts, activeEnter string }
+
+func (s edgeRunState) String() string { return s.restarts + "/" + s.activeEnter }
+
+// steady is the half of the soak that a comparison cannot make: a restart
+// that happened before the first sample is still a restart.
+func (s edgeRunState) steady() error {
+ if s.restarts != "0" {
+ return fmt.Errorf("edge has crash-restarted since the install (NRestarts=%s)", s.restarts)
+ }
+ return nil
+}
+
+func (a *systemApplier) edgeRunState() (edgeRunState, error) {
+ out, err := a.ops.systemctl("show", "-p", "NRestarts", "-p", "ActiveEnterTimestampMonotonic", "--value", systemEdgeUnit)
+ if err != nil {
+ return edgeRunState{}, err
+ }
+ restarts, rest, _ := strings.Cut(strings.TrimSpace(out), "\n")
+ return edgeRunState{restarts: strings.TrimSpace(restarts), activeEnter: strings.TrimSpace(rest)}, nil
+}
+
func (a *systemApplier) gateOnce(ctx context.Context, target string) error {
if out, err := a.ops.systemctl("is-active", systemEdgeUnit); err != nil || strings.TrimSpace(out) != "active" {
+ if err != nil {
+ return fmt.Errorf("%s is %s: %w", systemEdgeUnit, strings.TrimSpace(out), err)
+ }
return fmt.Errorf("%s is %s", systemEdgeUnit, strings.TrimSpace(out))
}
if err := a.ops.probeEdge(); err != nil {
@@ -435,8 +489,8 @@ func (a *systemApplier) rollback(ctx context.Context, req systemRequest, reason
return errors.New("rollback not implemented: " + reason)
}
-// diff is have − drop, order preserved.
-func diff(have, drop []string) []string {
+// unitsMinus is have − drop, order preserved.
+func unitsMinus(have, drop []string) []string {
skip := map[string]bool{}
for _, d := range drop {
skip[d] = true
@@ -451,49 +505,17 @@ func diff(have, drop []string) []string {
}
// copyFileDurable is writeFileDurable for a file too large to hold in
-// memory: temp, fsync, rename, directory fsync.
-func copyFileDurable(src, dst string, mode os.FileMode) (err error) {
+// memory — the binary itself — and lands the same way.
+func copyFileDurable(src, dst string, mode os.FileMode) error {
in, err := os.Open(src)
if err != nil {
return err
}
defer in.Close()
- tmp := dst + ".tmp"
- out, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, mode)
- if err != nil {
- return err
- }
- defer func() {
- if err != nil {
- out.Close()
- os.Remove(tmp)
- }
- }()
- if _, err = io.Copy(out, in); err != nil {
- return err
- }
- if err = out.Chmod(mode); err != nil {
- return err
- }
- if err = out.Sync(); err != nil {
+ return writeDurable(dst, mode, func(w io.Writer) error {
+ _, err := io.Copy(w, in)
return err
- }
- if err = out.Close(); err != nil {
- return err
- }
- if err = os.Rename(tmp, dst); err != nil {
- return err
- }
- return syncDirBestEffort(filepath.Dir(dst))
-}
-
-func syncDirBestEffort(dir string) error {
- d, err := os.Open(dir)
- if err != nil {
- return err
- }
- defer d.Close()
- return d.Sync()
+ })
}
func (a *systemApplier) recoverIfNeeded(ctx context.Context) error {
diff --git a/internal/host/systemapply_test.go b/internal/host/systemapply_test.go
index 090a20600..d3b8cd4f0 100644
--- a/internal/host/systemapply_test.go
+++ b/internal/host/systemapply_test.go
@@ -393,6 +393,9 @@ func applyHappy(t *testing.T) (*applyFixture, systemRequest) {
f.answer["version:"+staged] = "carlos v2\n"
f.answer["plan:"+staged] = `{"enabled":["carlos-edge.service","carlos-adopt.timer","carlos-adopt.path","carlos-stage.path","carlos-restart.path","carlos-system-update.path","carlos-system-recover.service","carlos-new-only.path"],"inert":["carlos-restore-verify.timer"]}`
f.answer["units:"+staged] = "carlos-edge.service carlos-adopt.timer carlos-adopt.path carlos-stage.path carlos-restart.path carlos-system-update.path carlos-system-update.service carlos-system-recover.service carlos-new-only.path"
+ // The unit v1 owns and v2 drops is on the box, so removing it is a
+ // thing that can be seen to happen — or not to.
+ os.WriteFile(filepath.Join(f.a.etcDir, "systemd", "system", "carlos-old-only.timer"), []byte("[Unit]\n# v1\n"), 0o644)
// .prev, once made, answers as v1 and stages v1's units.
prev := systemPrevPath(f.a.optDir)
f.answer["version:"+prev] = "carlos v1\n"
@@ -445,7 +448,7 @@ func TestApplyInstallsGatesArmsAndRecords(t *testing.T) {
if _, err := os.Stat(filepath.Join(f.a.etcDir, "tmpfiles.d", "carlos.conf")); err != nil {
t.Error("tmpfiles not installed")
}
- if _, err := os.Stat(filepath.Join(f.a.etcDir, "registry.db")); err == nil {
+ if _, err := os.Stat(filepath.Join(f.a.etcDir, "data", "registry.db")); err == nil {
t.Error("staging data/ copied")
}
// Order of the systemctl calls is the spec.
@@ -540,4 +543,83 @@ func TestApplyPrevUnitsComeFromPrevNotFromDisk(t *testing.T) {
if !found {
t.Fatalf("prev units not staged from carlos.prev:\n%s", strings.Join(f.calls, "\n"))
}
+ // And units_removed follows from that set: v0's timer goes, and the
+ // on-disk v1's — a build nothing here is restoring — is left alone.
+ if !f.called("systemctl disable --now carlos-v0-only.timer") {
+ t.Errorf("carlos-v0-only.timer not removed:\n%s", strings.Join(f.calls, "\n"))
+ }
+ if f.called("systemctl disable --now carlos-old-only.timer") {
+ t.Errorf("removed a unit computed against the on-disk build:\n%s", strings.Join(f.calls, "\n"))
+ }
+}
+
+func TestApplyPrevIsRefreshedWhenAppliedNamesTheDisk(t *testing.T) {
+ // The ordinary box: applied.json names the binary that is on disk, so
+ // that binary passed a gate here and is the one to fall back to.
+ f, _ := applyHappy(t)
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v0\n"), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.a.optDir), "applied.json"), systemApplied{Version: "v1", Channel: "stable", At: f.now}, 0o644)
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ prev, _ := os.ReadFile(systemPrevPath(f.a.optDir))
+ if !strings.Contains(string(prev), "carlos v1") {
+ t.Fatalf(".prev = %q, want the v1 the box had applied", prev)
+ }
+}
+
+func TestApplyPrevIsLeftAloneWhenNothingWasEverApplied(t *testing.T) {
+ // No applied.json, but a .prev exists: the box's history is unknown, so
+ // the untested disk binary does not get to become the fallback.
+ f, _ := applyHappy(t)
+ os.WriteFile(systemPrevPath(f.a.optDir), []byte("#!/bin/sh\necho carlos v0\n"), 0o755)
+ f.answer["version:"+systemPrevPath(f.a.optDir)] = "carlos v0\n"
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ prev, _ := os.ReadFile(systemPrevPath(f.a.optDir))
+ if !strings.Contains(string(prev), "carlos v0") {
+ t.Fatalf(".prev = %q, want the v0 that was already there", prev)
+ }
+}
+
+func TestApplyGateDemandsZeroRestartsNotJustAStillOne(t *testing.T) {
+ // The edge crashed once and then settled: nothing moves across the
+ // soak, but NRestarts is not 0 and the build is not one to keep.
+ f, req := applyHappy(t)
+ f.nrestarts = "1"
+ if err := f.a.Run(context.Background(), false); err == nil {
+ t.Fatal("an edge that had already crash-restarted passed the gate")
+ }
+ r := f.result()
+ if r == nil || !r.matches("stable", "v2", req.PromotedAt) || r.Outcome == systemOutcomeApplied {
+ t.Fatalf("result = %+v", r)
+ }
+}
+
+func TestApplyRecordsAnArmingFailureAgainstTheApply(t *testing.T) {
+ // The gate passed, so the roll stands; but `enable` is one batched
+ // call, and a box left unarmed must not look like a clean apply.
+ f, req := applyHappy(t)
+ orig := f.a.ops.systemctl
+ f.a.ops.systemctl = func(args ...string) (string, error) {
+ if args[0] == "enable" {
+ orig(args...)
+ return "", errors.New("Failed to enable unit carlos-new-only.path")
+ }
+ return orig(args...)
+ }
+ if err := f.a.Run(context.Background(), false); err != nil {
+ t.Fatal(err)
+ }
+ r := f.result()
+ if r == nil || r.Outcome != systemOutcomeApplied || !r.matches("stable", "v2", req.PromotedAt) {
+ t.Fatalf("result = %+v", r)
+ }
+ if !strings.Contains(r.Reason, "armed with errors") {
+ t.Fatalf("reason = %q, want the arming failure recorded", r.Reason)
+ }
+ if ap, _ := readSystemApplied(f.a.optDir); ap == nil || ap.Version != "v2" {
+ t.Fatalf("applied = %+v (the gate passed; the roll stands)", ap)
+ }
}
diff --git a/internal/host/systemstate.go b/internal/host/systemstate.go
index 1bef0ebe9..9f4548e3d 100644
--- a/internal/host/systemstate.go
+++ b/internal/host/systemstate.go
@@ -3,6 +3,7 @@ package host
import (
"encoding/json"
"errors"
+ "io"
"os"
"path/filepath"
"regexp"
@@ -125,10 +126,13 @@ func systemRootDir(optDir string) string { return filepath.Join(optDir, "syst
func systemPrevPath(optDir string) string { return filepath.Join(optDir, "carlos.prev") }
func systemBinaryPath(optDir string) string { return filepath.Join(optDir, "carlos") }
-// writeFileDurable: temp, fsync, rename, directory fsync. A crash leaves the
-// old file or the new one, never a torn one, and the rename itself is on
-// disk before the caller's next step depends on it.
-func writeFileDurable(path string, data []byte, mode os.FileMode) (err error) {
+// writeDurable is the one durable-write protocol on this box: fn fills
+// <path>.tmp, which is fsynced, renamed over path, and the directory
+// fsynced. A crash leaves the old file or the new one, never a torn one,
+// and the rename itself is on disk before the caller's next step depends
+// on it. A failure anywhere leaves the temp file behind no more than the
+// target: it is removed.
+func writeDurable(path string, mode os.FileMode, fn func(io.Writer) error) (err error) {
tmp := path + ".tmp"
f, err := os.OpenFile(tmp, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, mode)
if err != nil {
@@ -140,7 +144,7 @@ func writeFileDurable(path string, data []byte, mode os.FileMode) (err error) {
os.Remove(tmp)
}
}()
- if _, err = f.Write(data); err != nil {
+ if err = fn(f); err != nil {
return err
}
if err = f.Chmod(mode); err != nil {
@@ -155,7 +159,21 @@ func writeFileDurable(path string, data []byte, mode os.FileMode) (err error) {
if err = os.Rename(tmp, path); err != nil {
return err
}
- d, err := os.Open(filepath.Dir(path))
+ return syncDir(filepath.Dir(path))
+}
+
+func writeFileDurable(path string, data []byte, mode os.FileMode) error {
+ return writeDurable(path, mode, func(w io.Writer) error {
+ _, err := w.Write(data)
+ return err
+ })
+}
+
+// syncDir fsyncs a directory, so a rename into it is on disk. What a
+// failure means is the caller's to decide: after a recorded apply it is a
+// line in the log, before one it is a reason to stop.
+func syncDir(dir string) error {
+ d, err := os.Open(dir)
if err != nil {
return err
}