| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/ed25519" |
| 6 | "crypto/sha256" |
| 7 | "encoding/hex" |
| 8 | "errors" |
| 9 | "io" |
| 10 | "log" |
| 11 | "math/rand/v2" |
| 12 | "os" |
| 13 | "path/filepath" |
| 14 | "runtime" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | "github.com/carlosframework/platform/internal/version" |
| 19 | ) |
| 20 | |
| 21 | // systemLeg is the sandboxed half of the platform self-update (spec §4): |
| 22 | // notice the channel pointer, fetch and verify the artifact into /data, ask |
| 23 | // root. Two entry points so the spread has something to run it: observe on |
| 24 | // a serial move (and hourly), tick every 2 s with no store read. |
| 25 | type systemLeg struct { |
| 26 | store release.Store |
| 27 | pub ed25519.PublicKey |
| 28 | channel string |
| 29 | platform string |
| 30 | build string |
| 31 | dataDir string |
| 32 | optDir string |
| 33 | |
| 34 | now func() time.Time |
| 35 | jitter func(time.Duration) time.Duration |
| 36 | // fetchTo is release.FetchToFile bound to this leg's store and limit, |
| 37 | // behind a seam so a test can count downloads and make one fail. |
| 38 | fetchTo func(ctx context.Context, key, dst, wantSHA256 string) error |
| 39 | logf func(format string, args ...any) |
| 40 | |
| 41 | pending *systemPending |
| 42 | // backoff is ONE retry ladder for everything that can go wrong after the |
| 43 | // deadline — an unreachable store on the fetch, an unwritable /data on |
| 44 | // the request, and root's own "deferred" — because they are all the same |
| 45 | // failure from this loop's point of view: the work could not be done, and |
| 46 | // hammering it every 2 s is what must not happen. Base 2 s doubling to |
| 47 | // 5 min, cleared when the pointer moves or root answers. |
| 48 | backoff time.Duration |
| 49 | backoffUntil time.Time |
| 50 | lastTouch time.Time |
| 51 | lastObserve time.Time |
| 52 | // loggedNoArtifact is the version whose "not for this box" line has |
| 53 | // already been said. Without it the hourly re-observe would repeat it |
| 54 | // forever for a fleet that mixes architectures. |
| 55 | loggedNoArtifact string |
| 56 | |
| 57 | observeHook func() // test seam |
| 58 | } |
| 59 | |
| 60 | // systemPending is the one promote this box is working on: what observe |
| 61 | // decided, and when tick may act on it. |
| 62 | type systemPending struct { |
| 63 | version string |
| 64 | promotedAt time.Time |
| 65 | noop bool |
| 66 | artifact release.SystemArtifact |
| 67 | deadline time.Time |
| 68 | requested bool |
| 69 | } |
| 70 | |
| 71 | const systemObserveEvery = time.Hour |
| 72 | |
| 73 | func newSystemLeg(store release.Store, pub ed25519.PublicKey, getenv func(string) string, dataDir, optDir string) *systemLeg { |
| 74 | channel, invalid := systemChannel(getenv) |
| 75 | l := &systemLeg{ |
| 76 | store: store, pub: pub, channel: channel, |
| 77 | platform: runtime.GOOS + "-" + runtime.GOARCH, build: version.Build, |
| 78 | dataDir: dataDir, optDir: optDir, |
| 79 | now: time.Now, |
| 80 | jitter: func(spread time.Duration) time.Duration { return rand.N(spread + 1) }, |
| 81 | fetchTo: func(ctx context.Context, key, dst, wantSHA256 string) error { |
| 82 | return release.FetchToFile(ctx, store, key, dst, systemArtifactLimit, wantSHA256) |
| 83 | }, |
| 84 | logf: log.Printf, |
| 85 | } |
| 86 | // Once, here at arming, rather than on every tick that reads the value. |
| 87 | if invalid != "" { |
| 88 | l.printf("system: %s=%q is not a channel; this box follows %s", systemChannelEnv, invalid, l.channel) |
| 89 | } |
| 90 | return l |
| 91 | } |
| 92 | |
| 93 | func (l *systemLeg) printf(format string, args ...any) { |
| 94 | if l.logf != nil { |
| 95 | l.logf(format, args...) |
| 96 | return |
| 97 | } |
| 98 | log.Printf(format, args...) |
| 99 | } |
| 100 | |
| 101 | // observe reads the pointer and decides. It never blocks on a download: |
| 102 | // that happens in tick, after the spread. |
| 103 | func (l *systemLeg) observe(ctx context.Context) { |
| 104 | if l.observeHook != nil { |
| 105 | l.observeHook() |
| 106 | return |
| 107 | } |
| 108 | now := l.now() |
| 109 | l.lastObserve = now |
| 110 | p, _, err := release.GetSystemPointer(ctx, l.store, l.pub, release.SystemComponentRouter, l.channel) |
| 111 | if err != nil { |
| 112 | if !errors.Is(err, os.ErrNotExist) { |
| 113 | l.printf("agent: system pointer %s: %v", l.channel, err) |
| 114 | } |
| 115 | return |
| 116 | } |
| 117 | seen, _ := readSystemSeen(l.optDir) |
| 118 | if wm, ok := seen[l.channel]; ok && p.PromotedAt.Before(wm) { |
| 119 | return // root would refuse it as a replay |
| 120 | } |
| 121 | if res, _ := readSystemResult(l.optDir); res.matches(l.channel, p.Current.Version, p.PromotedAt) && res.terminal() { |
| 122 | l.forget(p.Current.Version) |
| 123 | return // refuse to repeat a failure |
| 124 | } |
| 125 | noop := p.Current.Version == l.build |
| 126 | if noop { |
| 127 | if wm, ok := seen[l.channel]; ok && !p.PromotedAt.After(wm) { |
| 128 | return // nothing to do, and root has seen this promote |
| 129 | } |
| 130 | } |
| 131 | art, ok := p.Current.Artifacts[l.platform] |
| 132 | if !ok && !noop { |
| 133 | if l.loggedNoArtifact != p.Current.Version { |
| 134 | l.printf("agent: system pointer %s names %s with no artifact for %s — not for this box", l.channel, p.Current.Version, l.platform) |
| 135 | l.loggedNoArtifact = p.Current.Version |
| 136 | } |
| 137 | return |
| 138 | } |
| 139 | if l.pending != nil && l.pending.version == p.Current.Version && l.pending.promotedAt.Equal(p.PromotedAt) { |
| 140 | return // already pending; keep its deadline |
| 141 | } |
| 142 | // A different pointer is new work, not a continuation of the old one: |
| 143 | // its retry ladder starts at the base (same rule as noteRestartFailed's |
| 144 | // "a DIFFERENT version starts the hold fresh"). |
| 145 | l.backoff, l.backoffUntil = 0, time.Time{} |
| 146 | l.pending = &systemPending{version: p.Current.Version, promotedAt: p.PromotedAt, noop: noop, artifact: art, |
| 147 | deadline: now.Add(l.jitter(p.SpreadDuration()))} |
| 148 | if noop { |
| 149 | // Nothing to install and nothing to spread: root only needs to |
| 150 | // record the watermark, so it may as well hear about it now. |
| 151 | l.pending.deadline = now |
| 152 | } |
| 153 | l.pruneCache(p.Current.Version) |
| 154 | } |
| 155 | |
| 156 | // tick fires a pending request whose deadline has passed and re-touches an |
| 157 | // unanswered one every minute. It reads no store except for the one fetch. |
| 158 | func (l *systemLeg) tick(ctx context.Context, now time.Time) { |
| 159 | switch { |
| 160 | case l.lastObserve.IsZero(): |
| 161 | // First tick after arming: start the hourly clock rather than |
| 162 | // observe. The serial poll above is what makes a fresh promote |
| 163 | // early; this floor is only the backstop for a promote that |
| 164 | // landed while nobody bumped the serial. |
| 165 | l.lastObserve = now |
| 166 | case now.Sub(l.lastObserve) >= systemObserveEvery: |
| 167 | l.observe(ctx) |
| 168 | } |
| 169 | p := l.pending |
| 170 | if p == nil { |
| 171 | return |
| 172 | } |
| 173 | if res, _ := readSystemResult(l.optDir); res.matches(l.channel, p.version, p.promotedAt) { |
| 174 | switch { |
| 175 | case res.terminal(): |
| 176 | l.forget(p.version) |
| 177 | return |
| 178 | case res.Outcome == systemOutcomeNoted: |
| 179 | l.pending = nil |
| 180 | return |
| 181 | case res.Outcome == systemOutcomeDeferred && res.At.After(p.deadline): |
| 182 | // root could not reach the store; try again with backoff |
| 183 | p.deadline = now.Add(l.nextBackoff()) |
| 184 | p.requested = false |
| 185 | } |
| 186 | } |
| 187 | if now.Before(p.deadline) || now.Before(l.backoffUntil) { |
| 188 | return |
| 189 | } |
| 190 | if !p.requested { |
| 191 | if !p.noop { |
| 192 | if err := l.fetch(ctx, p); err != nil { |
| 193 | l.printf("agent: system update %s: %v", p.version, err) |
| 194 | os.Remove(systemCachePath(l.dataDir, p.version)) |
| 195 | l.holdOff(now) |
| 196 | return |
| 197 | } |
| 198 | } |
| 199 | if err := l.writeRequest(p, now); err != nil { |
| 200 | // Same ladder as the fetch: without it the next tick would |
| 201 | // re-run the whole download for a /data this process cannot |
| 202 | // write, every 2 seconds. |
| 203 | l.printf("agent: system update %s: request: %v", p.version, err) |
| 204 | l.holdOff(now) |
| 205 | return |
| 206 | } |
| 207 | p.requested = true |
| 208 | } |
| 209 | if now.Sub(l.lastTouch) >= systemRequestEvery { |
| 210 | if err := os.WriteFile(filepath.Join(l.dataDir, systemRequestMarker), []byte(now.UTC().Format(time.RFC3339Nano)+"\n"), 0o644); err != nil { |
| 211 | l.printf("agent: system update %s: cannot ask for a root apply: %v", p.version, err) |
| 212 | return |
| 213 | } |
| 214 | l.lastTouch = now |
| 215 | } |
| 216 | } |
| 217 | |
| 218 | // nextBackoff advances the retry ladder one rung: base 2 s, doubling, capped |
| 219 | // at 5 min — the clamp shape fastDeps.noteUnstaged uses, so a step can never |
| 220 | // overshoot the cap. |
| 221 | func (l *systemLeg) nextBackoff() time.Duration { |
| 222 | if l.backoff == 0 { |
| 223 | l.backoff = fastBackoffBase |
| 224 | } else if l.backoff *= 2; l.backoff > fastBackoffCap { |
| 225 | l.backoff = fastBackoffCap |
| 226 | } |
| 227 | return l.backoff |
| 228 | } |
| 229 | |
| 230 | // holdOff pushes the next attempt out by the ladder's next rung. What it |
| 231 | // bounds is a 96 MB download: a permanently unfetchable artifact must cost |
| 232 | // this box four attempts an hour, not one every two seconds. |
| 233 | func (l *systemLeg) holdOff(now time.Time) { l.backoffUntil = now.Add(l.nextBackoff()) } |
| 234 | |
| 235 | func (l *systemLeg) fetch(ctx context.Context, p *systemPending) error { |
| 236 | dst := systemCachePath(l.dataDir, p.version) |
| 237 | if systemCacheHolds(dst, p.artifact.SHA256) { |
| 238 | return nil // already downloaded and verified — a deferred retry or a |
| 239 | // re-promote of the same version must not pull it again |
| 240 | } |
| 241 | if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil { |
| 242 | return err |
| 243 | } |
| 244 | return l.fetchTo(ctx, p.artifact.Key, dst, p.artifact.SHA256) |
| 245 | } |
| 246 | |
| 247 | // systemCacheHolds reports whether the cache file at path IS the artifact |
| 248 | // the pointer signed. It hashes the file: presence proves nothing, and a |
| 249 | // truncated or swapped file is exactly what the signed sha is there for. |
| 250 | func systemCacheHolds(path, wantSHA256 string) bool { |
| 251 | if wantSHA256 == "" { |
| 252 | return false |
| 253 | } |
| 254 | f, err := os.Open(path) |
| 255 | if err != nil { |
| 256 | return false |
| 257 | } |
| 258 | defer f.Close() |
| 259 | h := sha256.New() |
| 260 | if _, err := io.Copy(h, io.LimitReader(f, systemArtifactLimit+1)); err != nil { |
| 261 | return false |
| 262 | } |
| 263 | return hex.EncodeToString(h.Sum(nil)) == wantSHA256 |
| 264 | } |
| 265 | |
| 266 | func (l *systemLeg) writeRequest(p *systemPending, now time.Time) error { |
| 267 | if err := os.MkdirAll(systemDataDir(l.dataDir), 0o755); err != nil { |
| 268 | return err |
| 269 | } |
| 270 | return writeJSONDurable(systemRequestPath(l.dataDir), systemRequest{Version: p.version, PromotedAt: p.promotedAt, NoOp: p.noop, SeenAt: now.UTC()}, 0o644) |
| 271 | } |
| 272 | |
| 273 | // pruneCache drops every cached version except keep. A promote that moves |
| 274 | // v2 → v3 before root answered would otherwise leave v2's binary under /data |
| 275 | // for good — nobody else deletes there. Only names validSystemVersion |
| 276 | // accepts are removed, so requested.json and anything else under the |
| 277 | // directory cannot be caught by it. |
| 278 | func (l *systemLeg) pruneCache(keep string) { |
| 279 | dir := systemDataDir(l.dataDir) |
| 280 | ents, err := os.ReadDir(dir) |
| 281 | if err != nil { |
| 282 | return |
| 283 | } |
| 284 | for _, e := range ents { |
| 285 | if !e.IsDir() || e.Name() == keep || !validSystemVersion(e.Name()) { |
| 286 | continue |
| 287 | } |
| 288 | if err := os.RemoveAll(filepath.Join(dir, e.Name())); err != nil { |
| 289 | l.printf("agent: system update: cannot drop the cached %s: %v", e.Name(), err) |
| 290 | } |
| 291 | } |
| 292 | } |
| 293 | |
| 294 | // forget: root has answered with a terminal outcome. Drop the pending item |
| 295 | // and the cache — root never deletes under /data, the agent does. Only the |
| 296 | // pending for THIS version is dropped: a newer pointer may already be armed. |
| 297 | func (l *systemLeg) forget(version string) { |
| 298 | if l.pending != nil && l.pending.version == version { |
| 299 | l.pending = nil |
| 300 | l.backoff, l.backoffUntil = 0, time.Time{} |
| 301 | } |
| 302 | if validSystemVersion(version) { |
| 303 | os.RemoveAll(filepath.Dir(systemCachePath(l.dataDir, version))) |
| 304 | } |
| 305 | } |
| 306 | |
amadan