amadan

carlos / platform Public

host: the agent notices a platform promote, fetches, verifies, and asks root

observe on a serial move (and hourly), tick every 2 s: spread jitter,
streamed fetch with the signed sha, a request carrying only version
and promoted_at, and a marker re-touched every minute until root
answers. A pointer root already rolled back is never re-requested.
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 c96a74089931bd111b4ffbae610955d9a9a52140 parent ff89246
3 files changed, +478 −1
  • internal/host/agent.go +28 −1
  • internal/host/systemleg.go +201 −0
  • internal/host/systemleg_test.go +249 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 81b996fe5..9f509684f 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -602,6 +602,10 @@ func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps)
// systemctl verb on the same unit — and it cannot wake anything: it
// fires only after a try-restart RETURNED AN ERROR, which a stopped
// unit's exit-0 no-op never does.
+ //
+ // DO NOT add the platform self-update here either: the system leg
+ // lives in the fast loop and re-observes hourly on its own, so an
+ // hourly call here would only double the pointer read.
if err := release.Pass(ctx, reg, store, pub, optDir, restart,
release.BakeDir(dataDir()), release.OnStageBlocked(stageRequester(dataDir())),
release.RestartHibernating(restart),
@@ -1087,6 +1091,7 @@ type fastDeps struct {
fullPass func(ctx context.Context) // reconcile → adopt → config sync, or a test hook
restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
+ systemLeg *systemLeg // the platform self-update leg (spec §4); nil = not armed
lastSerial int64
seeded bool
lastFull time.Time
@@ -1366,11 +1371,28 @@ func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
if d.scheduleLeg != nil {
d.scheduleLeg(ctx)
}
+ // The platform self-update leg rides the move too, last: a
+ // `carlos system update` is a console action like any other, and
+ // this is what makes a promote noticed in seconds instead of at
+ // the leg's own hourly floor. It only READS the pointer and sets
+ // a deadline — the fetch and the request happen in the tick
+ // below, after the spread, so a serial move costs one Get.
+ if d.systemLeg != nil {
+ d.systemLeg.observe(ctx)
+ }
d.lastSerial = doc.Serial
d.lastFull = now
}
}
+ // The self-update tick, outside the serial switch and before the sweep's
+ // early return: it is what runs a spread deadline down and re-touches an
+ // unanswered request, and neither has anything to do with routes. It
+ // reads no store unless a deadline has actually come due.
+ if d.systemLeg != nil {
+ d.systemLeg.tick(ctx, now)
+ }
+
members := d.fastMembers(now)
if len(members) == 0 {
return // zero store reads: no PassFilter, no bake-clock write
@@ -1507,7 +1529,12 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// declarations. Nil when the box has no scheduled work armed, which
// fastTick tolerates.
d.scheduleLeg = scheduleLegFor(sd)
- log.Printf("agent: fast converge loop armed (2s + jitter: serial poll + canary/* + unstaged routes)")
+ // The platform self-update leg (spec §4): the sandboxed half, which only
+ // ever fetches into /data and writes a request. It lives here rather than
+ // in a loop of its own because the spread deadline needs a seconds
+ // cadence to run down on, and this is the loop that already has one.
+ d.systemLeg = newSystemLeg(store, pub, os.Getenv, dd, optDir)
+ log.Printf("agent: fast converge loop armed (2s + jitter: serial poll + canary/* + unstaged routes + system pointer)")
for {
time.Sleep(fastTickEvery + rand.N(time.Second))
fastTick(ctx, d, time.Now())
diff --git a/internal/host/systemleg.go b/internal/host/systemleg.go
new file mode 100644
index 000000000..8e0c70c90
--- /dev/null
+++ b/internal/host/systemleg.go
@@ -0,0 +1,201 @@
+package host
+
+import (
+ "context"
+ "crypto/ed25519"
+ "errors"
+ "log"
+ "math/rand/v2"
+ "os"
+ "path/filepath"
+ "runtime"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/version"
+)
+
+// systemLeg is the sandboxed half of the platform self-update (spec §4):
+// notice the channel pointer, fetch and verify the artifact into /data, ask
+// root. Two entry points so the spread has something to run it: observe on
+// a serial move (and hourly), tick every 2 s with no store read.
+type systemLeg struct {
+ store release.Store
+ pub ed25519.PublicKey
+ channel string
+ platform string
+ build string
+ dataDir string
+ optDir string
+
+ now func() time.Time
+ jitter func(time.Duration) time.Duration
+
+ pending *systemPending
+ lastTouch time.Time
+ lastObserve time.Time
+ backoffUntil time.Time
+ backoff time.Duration
+
+ observeHook func() // test seam
+}
+
+// systemPending is the one promote this box is working on: what observe
+// decided, and when tick may act on it.
+type systemPending struct {
+ version string
+ promotedAt time.Time
+ noop bool
+ artifact release.SystemArtifact
+ deadline time.Time
+ requested bool
+}
+
+const systemObserveEvery = time.Hour
+
+func newSystemLeg(store release.Store, pub ed25519.PublicKey, getenv func(string) string, dataDir, optDir string) *systemLeg {
+ return &systemLeg{
+ store: store, pub: pub, channel: systemChannel(getenv),
+ platform: runtime.GOOS + "-" + runtime.GOARCH, build: version.Build,
+ dataDir: dataDir, optDir: optDir,
+ now: time.Now,
+ jitter: func(spread time.Duration) time.Duration { return rand.N(spread + 1) },
+ }
+}
+
+// observe reads the pointer and decides. It never blocks on a download:
+// that happens in tick, after the spread.
+func (l *systemLeg) observe(ctx context.Context) {
+ if l.observeHook != nil {
+ l.observeHook()
+ return
+ }
+ now := l.now()
+ l.lastObserve = now
+ p, _, err := release.GetSystemPointer(ctx, l.store, l.pub, release.SystemComponentRouter, l.channel)
+ if err != nil {
+ if !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: system pointer %s: %v", l.channel, err)
+ }
+ return
+ }
+ seen, _ := readSystemSeen(l.optDir)
+ if wm, ok := seen[l.channel]; ok && p.PromotedAt.Before(wm) {
+ return // root would refuse it as a replay
+ }
+ if res, _ := readSystemResult(l.optDir); res.matches(l.channel, p.Current.Version, p.PromotedAt) && res.terminal() {
+ l.forget(p.Current.Version)
+ return // refuse to repeat a failure
+ }
+ noop := p.Current.Version == l.build
+ if noop {
+ if wm, ok := seen[l.channel]; ok && !p.PromotedAt.After(wm) {
+ return // nothing to do, and root has seen this promote
+ }
+ }
+ art, ok := p.Current.Artifacts[l.platform]
+ if !ok && !noop {
+ if l.pending == nil || l.pending.version != p.Current.Version {
+ log.Printf("agent: system pointer %s names %s with no artifact for %s — not for this box", l.channel, p.Current.Version, l.platform)
+ }
+ return
+ }
+ if l.pending != nil && l.pending.version == p.Current.Version && l.pending.promotedAt.Equal(p.PromotedAt) {
+ return // already pending; keep its deadline
+ }
+ l.pending = &systemPending{version: p.Current.Version, promotedAt: p.PromotedAt, noop: noop, artifact: art,
+ deadline: now.Add(l.jitter(p.SpreadDuration()))}
+ if noop {
+ // Nothing to install and nothing to spread: root only needs to
+ // record the watermark, so it may as well hear about it now.
+ l.pending.deadline = now
+ }
+}
+
+// tick fires a pending request whose deadline has passed and re-touches an
+// unanswered one every minute. It reads no store except for the one fetch.
+func (l *systemLeg) tick(ctx context.Context, now time.Time) {
+ switch {
+ case l.lastObserve.IsZero():
+ // First tick after arming: start the hourly clock rather than
+ // observe. The serial poll above is what makes a fresh promote
+ // early; this floor is only the backstop for a promote that
+ // landed while nobody bumped the serial.
+ l.lastObserve = now
+ case now.Sub(l.lastObserve) >= systemObserveEvery:
+ l.observe(ctx)
+ }
+ p := l.pending
+ if p == nil {
+ return
+ }
+ if res, _ := readSystemResult(l.optDir); res.matches(l.channel, p.version, p.promotedAt) {
+ switch {
+ case res.terminal():
+ l.forget(p.version)
+ return
+ case res.Outcome == systemOutcomeNoted:
+ l.pending = nil
+ return
+ case res.Outcome == systemOutcomeDeferred && res.At.After(p.deadline):
+ // root could not reach the store; try again with backoff
+ if l.backoff == 0 {
+ l.backoff = fastBackoffBase
+ } else if l.backoff < fastBackoffCap {
+ l.backoff *= 2
+ }
+ p.deadline = now.Add(l.backoff)
+ p.requested = false
+ }
+ }
+ if now.Before(p.deadline) || now.Before(l.backoffUntil) {
+ return
+ }
+ if !p.requested {
+ if !p.noop {
+ if err := l.fetch(ctx, p); err != nil {
+ log.Printf("agent: system update %s: %v", p.version, err)
+ os.Remove(systemCachePath(l.dataDir, p.version))
+ l.backoffUntil = now.Add(fastBackoffBase)
+ return
+ }
+ }
+ if err := l.writeRequest(p, now); err != nil {
+ log.Printf("agent: system update %s: request: %v", p.version, err)
+ return
+ }
+ p.requested = true
+ }
+ if now.Sub(l.lastTouch) >= systemRequestEvery {
+ if err := os.WriteFile(filepath.Join(l.dataDir, systemRequestMarker), []byte(now.UTC().Format(time.RFC3339Nano)+"\n"), 0o644); err != nil {
+ log.Printf("agent: system update %s: cannot ask for a root apply: %v", p.version, err)
+ return
+ }
+ l.lastTouch = now
+ }
+}
+
+func (l *systemLeg) fetch(ctx context.Context, p *systemPending) error {
+ dst := systemCachePath(l.dataDir, p.version)
+ if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
+ return err
+ }
+ return release.FetchToFile(ctx, l.store, p.artifact.Key, dst, systemArtifactLimit, p.artifact.SHA256)
+}
+
+func (l *systemLeg) writeRequest(p *systemPending, now time.Time) error {
+ if err := os.MkdirAll(systemDataDir(l.dataDir), 0o755); err != nil {
+ return err
+ }
+ return writeJSONDurable(systemRequestPath(l.dataDir), systemRequest{Version: p.version, PromotedAt: p.promotedAt, NoOp: p.noop, SeenAt: now.UTC()}, 0o644)
+}
+
+// forget: root has answered with a terminal outcome. Drop the pending item
+// and the cache — root never deletes under /data, the agent does.
+func (l *systemLeg) forget(version string) {
+ l.pending = nil
+ l.backoff = 0
+ if validSystemVersion(version) {
+ os.RemoveAll(filepath.Dir(systemCachePath(l.dataDir, version)))
+ }
+}
diff --git a/internal/host/systemleg_test.go b/internal/host/systemleg_test.go
new file mode 100644
index 000000000..437e74f3b
--- /dev/null
+++ b/internal/host/systemleg_test.go
@@ -0,0 +1,249 @@
+package host
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/sha256"
+ "encoding/hex"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type legFixture struct {
+ t *testing.T
+ store *release.DirStore
+ priv ed25519.PrivateKey
+ leg *systemLeg
+ dd string
+ opt string
+ now time.Time
+}
+
+func newLegFixture(t *testing.T, channel, build string) *legFixture {
+ t.Helper()
+ pub, priv, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatalf("generate key: %v", err)
+ }
+ f := &legFixture{t: t, store: release.NewDirStore(t.TempDir()), priv: priv, dd: t.TempDir(), opt: t.TempDir(),
+ now: time.Date(2026, 9, 6, 10, 0, 0, 0, time.UTC)}
+ getenv := func(k string) string {
+ if k == systemChannelEnv {
+ return channel
+ }
+ return ""
+ }
+ f.leg = newSystemLeg(f.store, pub, getenv, f.dd, f.opt)
+ f.leg.build = build
+ f.leg.platform = "linux-arm64"
+ f.leg.now = func() time.Time { return f.now }
+ f.leg.jitter = func(d time.Duration) time.Duration { return d / 2 } // deterministic: half the spread
+ return f
+}
+
+func (f *legFixture) promote(version string, body []byte, spread time.Duration) {
+ f.t.Helper()
+ sum := sha256.Sum256(body)
+ rel := release.SystemRelease{Version: version, Artifacts: map[string]release.SystemArtifact{
+ "linux-arm64": {Key: release.SystemArtifactKey(version, "linux-arm64"), SHA256: hex.EncodeToString(sum[:])}}}
+ if _, err := f.store.Put(context.Background(), rel.Artifacts["linux-arm64"].Key, body, ""); err != nil {
+ f.t.Fatal(err)
+ }
+ if _, err := release.PromoteSystem(context.Background(), f.store, release.RootSigner(f.priv), release.SystemComponentRouter, f.leg.channel, rel,
+ release.SystemPromoteOptions{Spread: spread}); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+// replaceArtifact swaps the object under the signed sha. DirStore.Put is
+// create-only with an empty if-match, so the overwrite carries the current
+// etag — the point is a store whose bytes no longer match the pointer.
+func (f *legFixture) replaceArtifact(version string, body []byte) {
+ f.t.Helper()
+ key := release.SystemArtifactKey(version, "linux-arm64")
+ _, etag, err := f.store.Get(context.Background(), key)
+ if err != nil {
+ f.t.Fatal(err)
+ }
+ if _, err := f.store.Put(context.Background(), key, body, etag); err != nil {
+ f.t.Fatal(err)
+ }
+}
+
+func (f *legFixture) marker() (os.FileInfo, error) {
+ return os.Stat(filepath.Join(f.dd, systemRequestMarker))
+}
+
+func (f *legFixture) request() *systemRequest {
+ var r systemRequest
+ if err := readJSONFile(systemRequestPath(f.dd), &r); err != nil {
+ return nil
+ }
+ return &r
+}
+
+func TestSystemLegDoesNothingWithoutAPointer(t *testing.T) {
+ f := newLegFixture(t, "stable", "v1")
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now.Add(time.Hour))
+ if _, err := f.marker(); err == nil {
+ t.Fatal("marker touched with no pointer")
+ }
+}
+
+func TestSystemLegFetchesVerifiesAndRequestsAfterTheSpread(t *testing.T) {
+ f := newLegFixture(t, "stable", "v1")
+ body := []byte("ELF v2")
+ f.promote("v2", body, 10*time.Minute)
+ f.leg.observe(context.Background())
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested before the spread elapsed")
+ }
+ f.leg.tick(context.Background(), f.now.Add(4*time.Minute)) // jitter = 5m
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested before the deadline")
+ }
+ f.leg.tick(context.Background(), f.now.Add(5*time.Minute+time.Second))
+ if _, err := f.marker(); err != nil {
+ t.Fatal("marker not touched after the deadline")
+ }
+ got, _ := os.ReadFile(systemCachePath(f.dd, "v2"))
+ if string(got) != "ELF v2" {
+ t.Fatalf("cache = %q", got)
+ }
+ r := f.request()
+ if r == nil || r.Version != "v2" || r.NoOp || r.PromotedAt.IsZero() {
+ t.Fatalf("request = %+v", r)
+ }
+ raw, _ := os.ReadFile(systemRequestPath(f.dd))
+ for _, never := range []string{"channel", "key", "sha", "path"} {
+ if strings.Contains(string(raw), `"`+never+`"`) {
+ t.Errorf("request carries %q — root must not take that from the sandbox", never)
+ }
+ }
+}
+
+func TestSystemLegRetouchesUntilRootAnswers(t *testing.T) {
+ f := newLegFixture(t, "stable", "v1")
+ f.promote("v2", []byte("x"), 0)
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now)
+ first, _ := f.marker()
+ f.leg.tick(context.Background(), f.now.Add(30*time.Second))
+ same, _ := f.marker()
+ if !same.ModTime().Equal(first.ModTime()) {
+ t.Fatal("re-touched inside the minute throttle")
+ }
+ // The touch is a rewrite (PathChanged needs a change), so mtime moves.
+ time.Sleep(20 * time.Millisecond)
+ f.leg.tick(context.Background(), f.now.Add(61*time.Second))
+ later, _ := f.marker()
+ if !later.ModTime().After(first.ModTime()) {
+ t.Fatal("not re-touched after a minute with no result")
+ }
+ // Root answers: a terminal result for this tuple stops the re-touch and
+ // the leg deletes its own cache.
+ r := f.request()
+ os.MkdirAll(systemRootDir(f.opt), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.opt), "result.json"),
+ systemResult{Channel: "stable", Version: "v2", PromotedAt: r.PromotedAt, Outcome: systemOutcomeRolledBack, Reason: "gate", At: f.now}, 0o644)
+ time.Sleep(20 * time.Millisecond)
+ f.leg.tick(context.Background(), f.now.Add(3*time.Minute))
+ final, _ := f.marker()
+ if !final.ModTime().Equal(later.ModTime()) {
+ t.Fatal("re-touched after a terminal result")
+ }
+ if _, err := os.Stat(systemCachePath(f.dd, "v2")); err == nil {
+ t.Fatal("cache kept after a terminal result")
+ }
+ // And a later observe of the SAME pointer does nothing: refuse to repeat a failure.
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now.Add(10*time.Minute))
+ if f.leg.pending != nil {
+ t.Fatal("re-requested a pointer root already rolled back")
+ }
+}
+
+func TestSystemLegNoOpRequestFetchesNothing(t *testing.T) {
+ f := newLegFixture(t, "stable", "v2")
+ f.promote("v2", []byte("same"), 0)
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now)
+ r := f.request()
+ if r == nil || !r.NoOp {
+ t.Fatalf("request = %+v, want a no-op", r)
+ }
+ if _, err := os.Stat(systemCachePath(f.dd, "v2")); err == nil {
+ t.Fatal("a no-op must not download the artifact")
+ }
+ // Once root has seen it (seen.json advanced to this PromotedAt), nothing more.
+ os.MkdirAll(systemRootDir(f.opt), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.opt), "seen.json"), systemSeen{"stable": r.PromotedAt}, 0o644)
+ f.leg.pending = nil
+ os.Remove(filepath.Join(f.dd, systemRequestMarker))
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now.Add(2*time.Minute))
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested again after root recorded the watermark")
+ }
+}
+
+func TestSystemLegIgnoresDevBuildEqualityAndOtherPlatforms(t *testing.T) {
+ f := newLegFixture(t, "stable", "dev")
+ f.promote("v2", []byte("x"), 0)
+ f.leg.platform = "linux-amd64" // no artifact for me
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now)
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested with no artifact for this platform")
+ }
+}
+
+func TestSystemLegRefusesShaMismatchAndOversize(t *testing.T) {
+ f := newLegFixture(t, "stable", "v1")
+ f.promote("v2", []byte("good"), 0)
+ // Swap the object under the signed sha.
+ f.replaceArtifact("v2", []byte("evil"))
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now)
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested an artifact whose sha does not match")
+ }
+ if _, err := os.Stat(systemCachePath(f.dd, "v2")); err == nil {
+ t.Fatal("mismatched artifact left in the cache")
+ }
+}
+
+func TestSystemLegIgnoresAPointerOlderThanTheWatermark(t *testing.T) {
+ f := newLegFixture(t, "stable", "v1")
+ f.promote("v2", []byte("x"), 0)
+ os.MkdirAll(systemRootDir(f.opt), 0o755)
+ writeJSONDurable(filepath.Join(systemRootDir(f.opt), "seen.json"), systemSeen{"stable": time.Now().Add(time.Hour)}, 0o644)
+ f.leg.observe(context.Background())
+ f.leg.tick(context.Background(), f.now)
+ if _, err := f.marker(); err == nil {
+ t.Fatal("requested a pointer root would refuse as a replay")
+ }
+}
+
+func TestFastTickRunsTheSystemLeg(t *testing.T) {
+ fx := newFastFixture(t)
+ d := fx.deps(nil)
+ observes := 0
+ d.systemLeg = &systemLeg{observeHook: func() { observes++ }}
+ ctx := context.Background()
+ fastTick(ctx, d, time.Now()) // boot seed
+ if observes != 0 {
+ t.Fatalf("boot seed observed %d times", observes)
+ }
+ release.BumpSerial(ctx, fx.store, time.Now())
+ fastTick(ctx, d, time.Now().Add(10*time.Second))
+ if observes != 1 {
+ t.Fatalf("one serial move → %d observes", observes)
+ }
+}