| index 8e0c70c90..ec4785a3b 100644 |
| --- a/internal/host/systemleg.go |
| +++ b/internal/host/systemleg.go |
| @@ -3,7 +3,10 @@ package host |
| import ( |
| "context" |
| "crypto/ed25519" |
| + "crypto/sha256" |
| + "encoding/hex" |
| "errors" |
| + "io" |
| "log" |
| "math/rand/v2" |
| "os" |
| @@ -30,12 +33,26 @@ type systemLeg struct { |
| |
| now func() time.Time |
| jitter func(time.Duration) time.Duration |
| + // fetchTo is release.FetchToFile bound to this leg's store and limit, |
| + // behind a seam so a test can count downloads and make one fail. |
| + fetchTo func(ctx context.Context, key, dst, wantSHA256 string) error |
| + logf func(format string, args ...any) |
| |
| - pending *systemPending |
| + pending *systemPending |
| + // backoff is ONE retry ladder for everything that can go wrong after the |
| + // deadline — an unreachable store on the fetch, an unwritable /data on |
| + // the request, and root's own "deferred" — because they are all the same |
| + // failure from this loop's point of view: the work could not be done, and |
| + // hammering it every 2 s is what must not happen. Base 2 s doubling to |
| + // 5 min, cleared when the pointer moves or root answers. |
| + backoff time.Duration |
| + backoffUntil time.Time |
| lastTouch time.Time |
| lastObserve time.Time |
| - backoffUntil time.Time |
| - backoff time.Duration |
| + // loggedNoArtifact is the version whose "not for this box" line has |
| + // already been said. Without it the hourly re-observe would repeat it |
| + // forever for a fleet that mixes architectures. |
| + loggedNoArtifact string |
| |
| observeHook func() // test seam |
| } |
| @@ -60,7 +77,19 @@ func newSystemLeg(store release.Store, pub ed25519.PublicKey, getenv func(string |
| dataDir: dataDir, optDir: optDir, |
| now: time.Now, |
| jitter: func(spread time.Duration) time.Duration { return rand.N(spread + 1) }, |
| + fetchTo: func(ctx context.Context, key, dst, wantSHA256 string) error { |
| + return release.FetchToFile(ctx, store, key, dst, systemArtifactLimit, wantSHA256) |
| + }, |
| + logf: log.Printf, |
| + } |
| +} |
| + |
| +func (l *systemLeg) printf(format string, args ...any) { |
| + if l.logf != nil { |
| + l.logf(format, args...) |
| + return |
| } |
| + log.Printf(format, args...) |
| } |
| |
| // observe reads the pointer and decides. It never blocks on a download: |
| @@ -75,7 +104,7 @@ func (l *systemLeg) observe(ctx context.Context) { |
| 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) |
| + l.printf("agent: system pointer %s: %v", l.channel, err) |
| } |
| return |
| } |
| @@ -95,14 +124,19 @@ func (l *systemLeg) observe(ctx context.Context) { |
| } |
| 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) |
| + if l.loggedNoArtifact != p.Current.Version { |
| + l.printf("agent: system pointer %s names %s with no artifact for %s — not for this box", l.channel, p.Current.Version, l.platform) |
| + l.loggedNoArtifact = p.Current.Version |
| } |
| return |
| } |
| if l.pending != nil && l.pending.version == p.Current.Version && l.pending.promotedAt.Equal(p.PromotedAt) { |
| return // already pending; keep its deadline |
| } |
| + // A different pointer is new work, not a continuation of the old one: |
| + // its retry ladder starts at the base (same rule as noteRestartFailed's |
| + // "a DIFFERENT version starts the hold fresh"). |
| + l.backoff, l.backoffUntil = 0, time.Time{} |
| l.pending = &systemPending{version: p.Current.Version, promotedAt: p.PromotedAt, noop: noop, artifact: art, |
| deadline: now.Add(l.jitter(p.SpreadDuration()))} |
| if noop { |
| @@ -110,6 +144,7 @@ func (l *systemLeg) observe(ctx context.Context) { |
| // record the watermark, so it may as well hear about it now. |
| l.pending.deadline = now |
| } |
| + l.pruneCache(p.Current.Version) |
| } |
| |
| // tick fires a pending request whose deadline has passed and re-touches an |
| @@ -139,12 +174,7 @@ func (l *systemLeg) tick(ctx context.Context, now time.Time) { |
| 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.deadline = now.Add(l.nextBackoff()) |
| p.requested = false |
| } |
| } |
| @@ -154,33 +184,77 @@ func (l *systemLeg) tick(ctx context.Context, now time.Time) { |
| if !p.requested { |
| if !p.noop { |
| if err := l.fetch(ctx, p); err != nil { |
| - log.Printf("agent: system update %s: %v", p.version, err) |
| + l.printf("agent: system update %s: %v", p.version, err) |
| os.Remove(systemCachePath(l.dataDir, p.version)) |
| - l.backoffUntil = now.Add(fastBackoffBase) |
| + l.holdOff(now) |
| return |
| } |
| } |
| if err := l.writeRequest(p, now); err != nil { |
| - log.Printf("agent: system update %s: request: %v", p.version, err) |
| + // Same ladder as the fetch: without it the next tick would |
| + // re-run the whole download for a /data this process cannot |
| + // write, every 2 seconds. |
| + l.printf("agent: system update %s: request: %v", p.version, err) |
| + l.holdOff(now) |
| 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) |
| + l.printf("agent: system update %s: cannot ask for a root apply: %v", p.version, err) |
| return |
| } |
| l.lastTouch = now |
| } |
| } |
| |
| +// nextBackoff advances the retry ladder one rung: base 2 s, doubling, capped |
| +// at 5 min — the clamp shape fastDeps.noteUnstaged uses, so a step can never |
| +// overshoot the cap. |
| +func (l *systemLeg) nextBackoff() time.Duration { |
| + if l.backoff == 0 { |
| + l.backoff = fastBackoffBase |
| + } else if l.backoff *= 2; l.backoff > fastBackoffCap { |
| + l.backoff = fastBackoffCap |
| + } |
| + return l.backoff |
| +} |
| + |
| +// holdOff pushes the next attempt out by the ladder's next rung. What it |
| +// bounds is a 96 MB download: a permanently unfetchable artifact must cost |
| +// this box four attempts an hour, not one every two seconds. |
| +func (l *systemLeg) holdOff(now time.Time) { l.backoffUntil = now.Add(l.nextBackoff()) } |
| + |
| func (l *systemLeg) fetch(ctx context.Context, p *systemPending) error { |
| dst := systemCachePath(l.dataDir, p.version) |
| + if systemCacheHolds(dst, p.artifact.SHA256) { |
| + return nil // already downloaded and verified — a deferred retry or a |
| + // re-promote of the same version must not pull it again |
| + } |
| 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) |
| + return l.fetchTo(ctx, p.artifact.Key, dst, p.artifact.SHA256) |
| +} |
| + |
| +// systemCacheHolds reports whether the cache file at path IS the artifact |
| +// the pointer signed. It hashes the file: presence proves nothing, and a |
| +// truncated or swapped file is exactly what the signed sha is there for. |
| +func systemCacheHolds(path, wantSHA256 string) bool { |
| + if wantSHA256 == "" { |
| + return false |
| + } |
| + f, err := os.Open(path) |
| + if err != nil { |
| + return false |
| + } |
| + defer f.Close() |
| + h := sha256.New() |
| + if _, err := io.Copy(h, io.LimitReader(f, systemArtifactLimit+1)); err != nil { |
| + return false |
| + } |
| + return hex.EncodeToString(h.Sum(nil)) == wantSHA256 |
| } |
| |
| func (l *systemLeg) writeRequest(p *systemPending, now time.Time) error { |
| @@ -190,11 +264,35 @@ func (l *systemLeg) writeRequest(p *systemPending, now time.Time) error { |
| return writeJSONDurable(systemRequestPath(l.dataDir), systemRequest{Version: p.version, PromotedAt: p.promotedAt, NoOp: p.noop, SeenAt: now.UTC()}, 0o644) |
| } |
| |
| +// pruneCache drops every cached version except keep. A promote that moves |
| +// v2 → v3 before root answered would otherwise leave v2's binary under /data |
| +// for good — nobody else deletes there. Only names validSystemVersion |
| +// accepts are removed, so requested.json and anything else under the |
| +// directory cannot be caught by it. |
| +func (l *systemLeg) pruneCache(keep string) { |
| + dir := systemDataDir(l.dataDir) |
| + ents, err := os.ReadDir(dir) |
| + if err != nil { |
| + return |
| + } |
| + for _, e := range ents { |
| + if !e.IsDir() || e.Name() == keep || !validSystemVersion(e.Name()) { |
| + continue |
| + } |
| + if err := os.RemoveAll(filepath.Join(dir, e.Name())); err != nil { |
| + l.printf("agent: system update: cannot drop the cached %s: %v", e.Name(), err) |
| + } |
| + } |
| +} |
| + |
| // forget: root has answered with a terminal outcome. Drop the pending item |
| -// and the cache — root never deletes under /data, the agent does. |
| +// and the cache — root never deletes under /data, the agent does. Only the |
| +// pending for THIS version is dropped: a newer pointer may already be armed. |
| func (l *systemLeg) forget(version string) { |
| - l.pending = nil |
| - l.backoff = 0 |
| + if l.pending != nil && l.pending.version == version { |
| + l.pending = nil |
| + l.backoff, l.backoffUntil = 0, time.Time{} |
| + } |
| if validSystemVersion(version) { |
| os.RemoveAll(filepath.Dir(systemCachePath(l.dataDir, version))) |
| } |
amadan