carlos / platform Public

zdd-6 task 10: the deploy watch follows each box's rollout and names promote --retry

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev bd27d3189a3c225cd377c6bb4d39392c60332eb0 parent ea06e4e
3 files changed, +1288 −132
  • deploy.go +122 −132
  • deploywatch.go +611 −0
  • deploywatch_test.go +555 −0
diff --git a/deploy.go b/deploy.go
index 56606ddef..8353510c6 100644
--- a/deploy.go
+++ b/deploy.go
@@ -15,7 +15,6 @@ import (
"io"
"log"
"net/http"
- "net/url"
"os"
"os/exec"
"path/filepath"
@@ -363,7 +362,7 @@ func runDeploy(args []string) {
}
fmt.Fprintf(stdout, "%s %s %s -> %s\n", colorFor(stdout).ok("promoted"), *app, *ver, channel)
if res.Held {
- mutedf(stdout, "%s is a production app: the %s bake window applies — the fleet adopts when it expires; console fast-track if this is an emergency\n", *app, channel)
+ mutedf(stdout, "%s\n", deployHeldLine(*app, channel, res.HeldUntil, time.Now()))
return // exit 0: the writes succeeded, the hold is policy (spec §6.6)
}
if len(hosts) == 0 {
@@ -410,56 +409,44 @@ func runDeploy(args []string) {
probe = ""
}
}
- if probe != "" {
- mutedf(stdout, "watching %s for the build %s reports itself\n", probe, *app)
- } else {
- // Said out loud, every time, because the silence is what the bug was
- // made of (discussion #13). With no probe this watch can only ask
- // the ROUTER what release was adopted; it cannot ask the process
- // what it is running. A deploy that goes green here has proved that
- // adoption happened, not that anything restarted — and an app whose
- // instance failed to cycle looks identical. It is a one-line fix
- // per app, so the line names it rather than describing the problem
- // and leaving the reader to find the verb.
- mutedf(stdout, "watching the X-Carlos-Version header: it reports the release ADOPTED, not the build running\n")
- // The advisory below assumes a process to probe. A static route has
- // none — no /api/version, no exec child to cycle — so the suggested
- // command would only teach the operator to configure a probe that
- // 404s, which then fails and times out every future `carlos deploy`
- // for that site. Static deploys get the line above (still true: the
- // header reports adoption, not a running build) and stop there.
- if !staticDeploy {
- mutedf(stdout, " a probe proves the process actually cycled — carlos apps build-probe --app %s /api/version\n", *app)
+ if staticDeploy {
+ // A static route has no box rollout to follow (no process, no
+ // scheduler), so it keeps the URL watch and its DNS guidance
+ // exactly as before.
+ printWatchNote(probe, true, *app)
+ targets := make([]watchTarget, 0, len(hosts))
+ for _, h := range hosts {
+ targets = append(targets, watchTarget{Host: h, Origin: watchOrigin(h), Probe: probe})
}
- }
- targets := make([]watchTarget, 0, len(hosts))
- for _, h := range hosts {
- targets = append(targets, watchTarget{Host: h, Origin: watchOrigin(h), Probe: probe})
- }
- diag := &watchDiagnostics{
- // A timed-out host's wake reason, from the platform log stream —
- // scoped tight (one host, errors only, the watch's own recent past)
- // so a diagnosis this may print was caused by THIS deploy, not an
- // incident from earlier in the day.
- wakeReason: func(host string) string {
- q := url.Values{"stream": {"platform"}, "level": {"error"},
- "host": {host}, "since": {"15m"}, "limit": {"1"}}
- res, err := sess.logsGet(ctx, acct, *app, q)
- if err != nil || len(res.Records) == 0 {
- return ""
+ if err := watchVersion(stdout, targets, *ver, *timeout, 2*time.Second, deployDiagnostics(ctx, sess, acct, *app)); err != nil {
+ if *hostFlag != "" && isConnectionClassFailure(err) {
+ failDeployDNSGuidance(ctx, sess, acct, *app, *hostFlag)
}
- return res.Records[len(res.Records)-1].Msg
- },
- logsHint: fmt.Sprintf("carlos logs --app %s --account %s --stream platform", *app, acct.Seg),
- }
- if err := watchVersion(stdout, targets, *ver, *timeout, 2*time.Second, diag); err != nil {
- if staticDeploy && *hostFlag != "" && isConnectionClassFailure(err) {
- failDeployDNSGuidance(ctx, sess, acct, *app, *hostFlag)
+ log.Fatalf("deploy: %v", err)
}
+ return
+ }
+ // An instance deploy follows each box's rollout of the build (spec §9.1,
+ // deploywatch.go), and is today's URL watch when no box reports one.
+ if err := watchDeployed(ctx, sess, acct, *app, *ver, channel, hosts, probe, *timeout); err != nil {
log.Fatalf("deploy: %v", err)
}
}
+// TODO-COPY zdd.deploy.held-until
+const deployHeldUntil = "%s is held by the %s bake window until about %s, then the fleet installs it. Fast-track it in the console if this is an emergency"
+
+// deployHeldLine is what `carlos deploy` says when the promote it made is
+// held by the channel's bake window (spec §6.6). The console's held_until,
+// when present and still ahead of now, is named in the operator's own zone
+// (R9); otherwise the line is today's, byte for byte.
+func deployHeldLine(app, channel, heldUntil string, now time.Time) string {
+ if t, err := time.Parse(time.RFC3339, heldUntil); err == nil && t.After(now) {
+ return fmt.Sprintf(deployHeldUntil, app, channel, t.Local().Format("Mon 2 Jan 15:04 MST"))
+ }
+ return fmt.Sprintf("%s is a production app: the %s bake window applies — the fleet adopts when it expires; console fast-track if this is an emergency", app, channel)
+}
+
// defaultDeployTimeout is how long --timeout watches a URL before failing,
// unless the operator says otherwise. It has to outlast one static recheck
// window with room to spare, or a static deploy that was always going to land
@@ -1496,93 +1483,8 @@ func watchVersion(w io.Writer, targets []watchTarget, ver string, timeout, inter
wg.Add(1)
go func(i int, t watchTarget) {
defer wg.Done()
- n := attempt.Add(1)
- // The cache-buster rides BOTH legs. The edge caches a proxied
- // 2xx — body included — for up to 5 minutes, longer than this
- // function's own default timeout, so an un-busted probe poll
- // could sit on one stale hit for the whole watch and report a
- // healthy deploy as failed.
- path := "/"
- if t.Probe != "" {
- path = t.Probe
- }
- url := fmt.Sprintf("%s%s?carlos-deploy-probe=%d", t.Origin, path, n)
- resp, err := client.Get(url)
- if err != nil {
- results[i] = result{diag: "unreachable: " + err.Error()}
- return
- }
- if t.Probe != "" {
- // The app's own process answers here, so the body IS the
- // truth — no header/route-row disagreement to reconcile
- // the way the header path has to. Read a bounded amount:
- // this is a health endpoint, and a target that answers
- // with a stream is misconfigured, not something to buffer.
- b, rerr := io.ReadAll(io.LimitReader(resp.Body, 4096))
- resp.Body.Close()
- if rerr != nil {
- results[i] = result{diag: "probe unreadable: " + rerr.Error()}
- return
- }
- got := strings.TrimSpace(string(b))
- if resp.StatusCode/100 != 2 {
- // Same rule as the header leg's 502 case: the right
- // build in a crash loop is not a live deploy.
- results[i] = result{diag: fmt.Sprintf("%s answered HTTP %d: %q", path, resp.StatusCode, got)}
- return
- }
- // Whole-token, never substring: abc123 must not satisfy a
- // watch for abc1234, and short git shas collide like that
- // in practice.
- for _, f := range strings.Fields(got) {
- if f == ver {
- results[i] = result{live: true}
- return
- }
- }
- // Naming the body is what makes the ldflags trap
- // self-diagnosing: a binary linked without
- // -X …/internal/version.Build reports "dev" forever, so a
- // perfectly healthy roll fails its watch — and "timed out"
- // would send the operator hunting the wrong thing, while
- // `reports "ok dev", waiting for abc1234` says what to fix.
- results[i] = result{diag: fmt.Sprintf("%s reports %q, waiting for %s", path, got, ver)}
- return
- }
- io.CopyN(io.Discard, resp.Body, 4096)
- resp.Body.Close()
- got := resp.Header.Get("X-Carlos-Version")
- switch {
- case got == ver && resp.StatusCode/100 == 2:
- results[i] = result{live: true}
- case got == ver:
- // The edge stamps X-Carlos-Version before proxying to the
- // origin, so a tenant that dies on boot can still answer
- // with the new version header attached to a non-2xx (a 502,
- // most commonly). That is the new build in a crash loop, not
- // a live deploy, so the success arm has to gate on status
- // too — matching the header alone let a dead tenant exit 0.
- results[i] = result{diag: fmt.Sprintf("serving %s but HTTP %d — the app is failing", ver, resp.StatusCode)}
- case got == "" && resp.StatusCode/100 != 2:
- // A non-2xx with no header is not an old edge — it is an edge
- // that answered and had nothing good to serve: an activator
- // 503 because the wake failed or timed out, or a 404 because
- // the route has not been minted yet. (A 502 from a tenant
- // that died on boot carries the version header — the edge
- // stamps it before proxying — so that shape lands in the
- // case above, not here.) Reporting those as "edge predates
- // this feature — the hourly pass will adopt it" sends the
- // operator off to wait for a pass that will not help (review
- // finding 2): the status code is the diagnostic, so print it.
- results[i] = result{diag: fmt.Sprintf("not serving (HTTP %d) — instance failing to wake, or route not live yet", resp.StatusCode)}
- case got == "":
- // 2xx and no header: something IS being served here, it just
- // isn't stamped — the only shape that really is an edge from
- // before X-Carlos-Version existed.
- results[i] = result{diag: "no X-Carlos-Version header (edge predates this feature — the hourly pass will adopt it)"}
- default:
- results[i] = result{diag: "still serving " + got}
- }
+ live, d := checkTarget(client, t, ver, attempt.Add(1))
+ results[i] = result{live: live, diag: d}
}(i, t)
}
wg.Wait()
@@ -1629,6 +1531,94 @@ func watchVersion(w io.Writer, targets []watchTarget, ver string, timeout, inter
}
}
+// checkTarget is one poll of one target: watchVersion's per-target body,
+// lifted out so the rollout watch (deploywatch.go) polls a host its box does
+// not report on exactly as this watch always has. n is the poll's
+// cache-buster, distinct for every poll of the watch.
+func checkTarget(client *http.Client, t watchTarget, ver string, n int64) (live bool, diag string) {
+ // The cache-buster rides BOTH legs. The edge caches a proxied
+ // 2xx — body included — for up to 5 minutes, longer than this
+ // function's own default timeout, so an un-busted probe poll
+ // could sit on one stale hit for the whole watch and report a
+ // healthy deploy as failed.
+ path := "/"
+ if t.Probe != "" {
+ path = t.Probe
+ }
+ url := fmt.Sprintf("%s%s?carlos-deploy-probe=%d", t.Origin, path, n)
+ resp, err := client.Get(url)
+ if err != nil {
+ return false, "unreachable: " + err.Error()
+ }
+ if t.Probe != "" {
+ // The app's own process answers here, so the body IS the
+ // truth — no header/route-row disagreement to reconcile
+ // the way the header path has to. Read a bounded amount:
+ // this is a health endpoint, and a target that answers
+ // with a stream is misconfigured, not something to buffer.
+ b, rerr := io.ReadAll(io.LimitReader(resp.Body, 4096))
+ resp.Body.Close()
+ if rerr != nil {
+ return false, "probe unreadable: " + rerr.Error()
+ }
+ got := strings.TrimSpace(string(b))
+ if resp.StatusCode/100 != 2 {
+ // Same rule as the header leg's 502 case: the right
+ // build in a crash loop is not a live deploy.
+ return false, fmt.Sprintf("%s answered HTTP %d: %q", path, resp.StatusCode, got)
+ }
+ // Whole-token, never substring: abc123 must not satisfy a
+ // watch for abc1234, and short git shas collide like that
+ // in practice.
+ for _, f := range strings.Fields(got) {
+ if f == ver {
+ return true, ""
+ }
+ }
+ // Naming the body is what makes the ldflags trap
+ // self-diagnosing: a binary linked without
+ // -X …/internal/version.Build reports "dev" forever, so a
+ // perfectly healthy roll fails its watch — and "timed out"
+ // would send the operator hunting the wrong thing, while
+ // `reports "ok dev", waiting for abc1234` says what to fix.
+ return false, fmt.Sprintf("%s reports %q, waiting for %s", path, got, ver)
+ }
+ io.CopyN(io.Discard, resp.Body, 4096)
+ resp.Body.Close()
+ got := resp.Header.Get("X-Carlos-Version")
+ switch {
+ case got == ver && resp.StatusCode/100 == 2:
+ return true, ""
+ case got == ver:
+ // The edge stamps X-Carlos-Version before proxying to the
+ // origin, so a tenant that dies on boot can still answer
+ // with the new version header attached to a non-2xx (a 502,
+ // most commonly). That is the new build in a crash loop, not
+ // a live deploy, so the success arm has to gate on status
+ // too — matching the header alone let a dead tenant exit 0.
+ return false, fmt.Sprintf("serving %s but HTTP %d — the app is failing", ver, resp.StatusCode)
+ case got == "" && resp.StatusCode/100 != 2:
+ // A non-2xx with no header is not an old edge — it is an edge
+ // that answered and had nothing good to serve: an activator
+ // 503 because the wake failed or timed out, or a 404 because
+ // the route has not been minted yet. (A 502 from a tenant
+ // that died on boot carries the version header — the edge
+ // stamps it before proxying — so that shape lands in the
+ // case above, not here.) Reporting those as "edge predates
+ // this feature — the hourly pass will adopt it" sends the
+ // operator off to wait for a pass that will not help (review
+ // finding 2): the status code is the diagnostic, so print it.
+ return false, fmt.Sprintf("not serving (HTTP %d) — instance failing to wake, or route not live yet", resp.StatusCode)
+ case got == "":
+ // 2xx and no header: something IS being served here, it just
+ // isn't stamped — the only shape that really is an edge from
+ // before X-Carlos-Version existed.
+ return false, "no X-Carlos-Version header (edge predates this feature — the hourly pass will adopt it)"
+ default:
+ return false, "still serving " + got
+ }
+}
+
// The domain lifecycle states the door derives (#299 §2a). Mirrored here as
// constants rather than compared as string literals so a typo is a build
// error; the door is the authority on the vocabulary and publishes it in the
diff --git a/deploywatch.go b/deploywatch.go
new file mode 100644
index 000000000..db3f3c3ca
--- /dev/null
+++ b/deploywatch.go
@@ -0,0 +1,611 @@
+package main
+
+// deploywatch.go is `carlos deploy`'s watch once boxes report their rollouts
+// (zero-downtime spec §9.1): completion is tied to the build being watched,
+// instance by instance, box by box. A host its box reports on is watched
+// through that report and never requested over HTTP, because a request would
+// wake an asleep instance the spec counts as done once its box has installed
+// the build. A host no box reports on is watched as it always was
+// (checkTarget). With no report at all, the watch IS watchVersion.
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/url"
+ "sort"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/rolloutwords"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+const (
+ // TODO-COPY zdd.watch.intro
+ watchIntro = "watching %s roll out box by box: an awake instance counts once the new build has taken over, an asleep one once its box has installed it"
+ // TODO-COPY zdd.watch.word-swapped
+ watchWordSwapped = "swapped"
+ // TODO-COPY zdd.watch.word-installed
+ watchWordInstalled = "installed"
+ // TODO-COPY zdd.watch.asleep-tail
+ watchAsleepTail = "starts on its next request"
+ // TODO-COPY zdd.watch.word-failed
+ watchWordFailed = "failed"
+ // TODO-COPY zdd.watch.word-held
+ watchWordHeld = "held"
+ // TODO-COPY zdd.watch.held-tail
+ watchHeldTail = "stayed on %s because %s failed on this box"
+ // TODO-COPY zdd.watch.box
+ watchBox = "box %s: %s"
+ // TODO-COPY zdd.watch.count-swapped
+ watchCountSwapped = "%d swapped"
+ // TODO-COPY zdd.watch.count-asleep
+ watchCountAsleep = "%d asleep with it installed"
+ // TODO-COPY zdd.watch.count-failed
+ watchCountFailed = "%d failed"
+ // TODO-COPY zdd.watch.count-held
+ watchCountHeld = "%d stayed on their last good build"
+ // TODO-COPY zdd.watch.failed-summary
+ watchFailedSummary = "build %s did not reach every instance: %d failed and %d stayed on their last good build. Fix it and deploy a new build, or try this one again: carlos promote --app %s --retry %s %s"
+ // TODO-COPY zdd.watch.hold-down-summary
+ watchHoldDownSummary = "an instance that swaps in hold mode is down, with nothing to fall back to: carlos rollback --app %s"
+ // TODO-COPY zdd.watch.timeout
+ watchTimeout = "timed out waiting for build %s:"
+ // TODO-COPY zdd.watch.pending-not-installed
+ watchPendingNotInstalled = "its box has not installed this build yet"
+ // TODO-COPY zdd.watch.pending-other
+ watchPendingOther = "its box has %s installed, not this build"
+ // TODO-COPY zdd.watch.pending-queued
+ watchPendingQueued = "queued"
+ // TODO-COPY zdd.watch.pending-running
+ watchPendingRunning = "starting beside the running build"
+ // TODO-COPY zdd.watch.pending-settling
+ watchPendingSettling = "settling: making sure it stays up for 10 seconds"
+ // TODO-COPY zdd.watch.pending-waiting
+ watchPendingWaiting = "waiting for its box to start it"
+ // TODO-COPY zdd.watch.pending-dark
+ watchPendingDark = "its box stopped reporting"
+ // TODO-COPY zdd.watch.pending-unlisted (new: no plan draft; controller ruling R3)
+ watchPendingUnlisted = "the console's instance list was cut short before it, so its state is unknown"
+ // TODO-COPY zdd.watch.box-busy
+ watchBoxBusy = "box %s is still working through this app's instances"
+)
+
+// deployWatchListEvery is how often the production watch reads the instance
+// list (controller ruling R5). The list is rate limited per token (60 a
+// minute, shared with the deploy's own calls), and each read costs the
+// console a records list and a box join, so it keeps its own cadence; a host
+// watched by URL is still polled every deployWatchPollEvery, as before.
+const (
+ deployWatchListEvery = 5 * time.Second
+ deployWatchPollEvery = 2 * time.Second
+)
+
+// rolloutState is one targeted instance's place in the rollout of the
+// watched build (spec §9.1): pending, or one of four final states.
+type rolloutState string
+
+const (
+ rolloutPending rolloutState = "pending"
+ rolloutSwapped rolloutState = "swapped"
+ rolloutAsleep rolloutState = "asleep" // installed, awaiting first wake
+ rolloutFailed rolloutState = "failed"
+ rolloutHeld rolloutState = "held"
+)
+
+func (s rolloutState) final() bool {
+ return s == rolloutSwapped || s == rolloutAsleep || s == rolloutFailed || s == rolloutHeld
+}
+
+// attemptOpen reports whether a box's attempt word is still in progress:
+// queued, running or settling, or a word this binary does not know (a newer
+// box), which is safer read as unfinished than as done. Failed and down are
+// outcomes, not attempts in progress.
+func attemptOpen(a string) bool {
+ return a != "" && a != status.AttemptFailed && a != status.AttemptDown
+}
+
+// classifyRollout decides one instance's state for build ver from its box's
+// report. In order:
+// - A box that cannot arm its scheduler (Unavailable) reports no per-host
+// state, so its awake=false means nothing: pending (R6).
+// - Until the box reports ver installed the instance is pending, however
+// idle its scheduler.
+// - A failure counts against ver only when the failed attempt was ver's,
+// and a hold-mode instance left down by ver is failed too (spec §9.1 "a
+// hold failure"). The console's join (Task 8) leaves the failed attempt
+// only on the host the build failed on, so this is checked before
+// held_by, which that host carries too.
+// - Held by ver, with no failure of its own: a sibling's failure record
+// holds it on its good build.
+// - An attempt in progress is pending whether or not the instance is awake
+// (R8): a canary serving ver has not swapped until its settle closes, and
+// an asleep instance whose first wake is under observation is not asleep.
+// - Serving ver with nothing in progress: swapped. Asleep with nothing in
+// progress: installed, awaiting first wake.
+func classifyRollout(r *apiRollout, ver string) rolloutState {
+ switch {
+ case r.Unavailable != "":
+ return rolloutPending
+ case r.Installed != ver:
+ return rolloutPending
+ case r.AttemptBuild == ver && (r.Attempt == status.AttemptFailed || r.Attempt == status.AttemptDown):
+ return rolloutFailed
+ case r.HeldBy == ver:
+ return rolloutHeld
+ case attemptOpen(r.Attempt):
+ return rolloutPending
+ case r.Serving == ver:
+ return rolloutSwapped
+ case !r.Awake && r.Serving == "" && r.Attempt == "":
+ return rolloutAsleep
+ }
+ return rolloutPending
+}
+
+// follows reports whether the watch follows an instance through r: a report
+// with per-host state, or one from a box that refuses every managed start
+// (its instances wait, and must not be woken by a poll). A box that declined
+// overlap runs today's stop-then-start paths, so its report is no report:
+// the instance is watched by URL, like a flag-off box's (R6).
+func follows(r *apiRollout) bool {
+ return r != nil && (r.Unavailable == "" || rolloutwords.UnavailableRefuses(r.Unavailable))
+}
+
+type watchMode int
+
+const (
+ modeUnknown watchMode = iota // not yet listed (a truncated list left it out)
+ modeRollout // watched through its box's report, for the rest of the watch
+ modeURL // watched by URL, as watchVersion does, until a report comes
+)
+
+type watchedHost struct {
+ t watchTarget
+ mode watchMode
+ // ro is the latest report the watch followed. It is kept when the box
+ // goes dark, so the host still counts on its box.
+ ro *apiRollout
+ // dark is set while the latest complete list carries no report for a
+ // host the watch follows (R1). It never sends the host to a URL poll.
+ dark bool
+ state rolloutState
+ // fin is the report the host went final on. A final state is final for
+ // the life of the watch (R2): what it printed and what it counts come
+ // from fin, never from a later report.
+ fin *apiRollout
+ urlLive bool
+ urlDiag string
+ printed bool
+}
+
+func (h *watchedHost) done() bool {
+ switch h.mode {
+ case modeURL:
+ return h.urlLive
+ case modeRollout:
+ return h.state.final()
+ }
+ return false
+}
+
+// rolloutWatch is one watch. list is the console's instance list (records,
+// whether the list was cut short, error); check is one URL poll (checkTarget
+// in production); onURL runs once, the first time any host is watched by
+// URL, so the probe-or-header note prints only when it is true. channel names
+// the `promote --retry` a failure suggests. interval is the URL poll cadence
+// and listEvery the list's (R5); a zero listEvery lists on every sweep.
+type rolloutWatch struct {
+ w io.Writer
+ app, ver, channel string
+ timeout, interval time.Duration
+ listEvery time.Duration
+ list func() ([]apiInstanceRecord, bool, error)
+ check func(t watchTarget, n int64) (bool, string)
+ onURL func()
+ diag *watchDiagnostics
+}
+
+func (rw rolloutWatch) run(targets []watchTarget) error {
+ deadline := time.Now().Add(rw.timeout)
+ recs, truncated, err := rw.list()
+ lastList := time.Now()
+ if err != nil || !reportsAny(recs, targets) {
+ // No box reports a rollout for any target: today's watch, byte for
+ // byte (R10).
+ if rw.onURL != nil {
+ rw.onURL()
+ }
+ return watchVersion(rw.w, targets, rw.ver, rw.timeout, rw.interval, rw.diag)
+ }
+ var hosts []*watchedHost
+ seen := map[string]bool{}
+ for _, t := range targets {
+ if !seen[t.Host] {
+ seen[t.Host] = true
+ hosts = append(hosts, &watchedHost{t: t, state: rolloutPending})
+ }
+ }
+ c := colorFor(rw.w)
+ mutedf(rw.w, watchIntro, rw.ver)
+ rw.apply(hosts, recs, truncated)
+ boxPrinted := map[string]bool{}
+ announcedURL := false
+ var n int64
+ for {
+ urlPending := false
+ for _, h := range hosts {
+ if h.mode == modeURL && !h.urlLive {
+ urlPending = true
+ }
+ }
+ if urlPending && !announcedURL {
+ announcedURL = true
+ if rw.onURL != nil {
+ rw.onURL()
+ }
+ }
+ var wg sync.WaitGroup
+ for _, h := range hosts {
+ if h.mode != modeURL || h.urlLive {
+ continue
+ }
+ n++
+ wg.Add(1)
+ go func(h *watchedHost, n int64) {
+ defer wg.Done()
+ h.urlLive, h.urlDiag = rw.check(h.t, n)
+ }(h, n)
+ }
+ wg.Wait()
+ for _, h := range hosts {
+ if h.done() && !h.printed {
+ h.printed = true
+ rw.printHost(c, h)
+ }
+ }
+ boxes, labels := groupByBox(hosts)
+ boxesDone := true
+ for _, l := range labels {
+ if !boxDone(boxes[l]) {
+ boxesDone = false
+ continue
+ }
+ if !boxPrinted[l] {
+ boxPrinted[l] = true
+ fmt.Fprintln(rw.w, boxSummary(l, boxes[l]))
+ }
+ }
+ hostsDone := true
+ for _, h := range hosts {
+ hostsDone = hostsDone && h.done()
+ }
+ if hostsDone && boxesDone {
+ return rw.result(hosts)
+ }
+ if time.Now().After(deadline) {
+ return rw.timedOut(hosts, boxes, labels)
+ }
+ time.Sleep(rw.wait(hosts, lastList, deadline))
+ if time.Since(lastList) >= rw.listEvery {
+ recs, truncated, err = rw.list()
+ lastList = time.Now()
+ // A list that failed (a rate limit, a console restart) changes
+ // nothing: every host keeps its mode and state until a list
+ // answers.
+ if err == nil {
+ rw.apply(hosts, recs, truncated)
+ }
+ }
+ }
+}
+
+// wait is how long the loop sleeps: the URL poll interval while any host is
+// watched by URL, otherwise until the next list is due; never past the
+// deadline, so the timeout report comes on time.
+func (rw rolloutWatch) wait(hosts []*watchedHost, lastList, deadline time.Time) time.Duration {
+ d := rw.interval
+ urlPending := false
+ for _, h := range hosts {
+ urlPending = urlPending || (h.mode == modeURL && !h.urlLive)
+ }
+ if !urlPending {
+ d = max(d, time.Until(lastList.Add(rw.listEvery)))
+ }
+ if left := time.Until(deadline); left > 0 && left < d {
+ d = left
+ }
+ return d
+}
+
+// reportsAny reports whether the list carries a report the watch follows for
+// any of targets. A report on an instance this deploy does not target says
+// nothing about the targets' boxes.
+func reportsAny(recs []apiInstanceRecord, targets []watchTarget) bool {
+ want := make(map[string]bool, len(targets))
+ for _, t := range targets {
+ want[t.Host] = true
+ }
+ for _, r := range recs {
+ if want[r.Host] && follows(r.Rollout) {
+ return true
+ }
+ }
+ return false
+}
+
+// apply folds one answered list into the hosts.
+//
+// - A host missing from a truncated list keeps its mode and state (R3):
+// the cap cut it off, nothing about it changed, and switching it to URL
+// polling would wake it if it is asleep.
+// - A host with a report the watch follows is watched through it from now
+// on (R4: also when it was being polled by URL). Its report is refreshed
+// every list even once it is final, because its box is done only on the
+// box's latest scheduler word; its state is not (R2).
+// - A host the watch follows whose report is gone is dark (R1): pending if
+// it was not final, never polled.
+// - Any other host is watched by URL, exactly as watchVersion would.
+func (rw rolloutWatch) apply(hosts []*watchedHost, recs []apiInstanceRecord, truncated bool) {
+ listed := make(map[string]*apiRollout, len(recs))
+ for i := range recs {
+ listed[recs[i].Host] = recs[i].Rollout
+ }
+ for _, h := range hosts {
+ ro, ok := listed[h.t.Host]
+ switch {
+ case !ok && truncated:
+ case h.mode == modeURL && h.urlLive:
+ // Live by URL is final too.
+ case follows(ro):
+ h.mode, h.ro, h.dark = modeRollout, ro, false
+ if !h.state.final() {
+ h.state = classifyRollout(ro, rw.ver)
+ if h.state.final() {
+ fin := *ro
+ h.fin = &fin
+ }
+ }
+ case h.mode == modeRollout:
+ h.dark = true
+ default:
+ h.mode = modeURL
+ }
+ }
+}
+
+func groupByBox(hosts []*watchedHost) (map[string][]*watchedHost, []string) {
+ boxes := map[string][]*watchedHost{}
+ var labels []string
+ for _, h := range hosts {
+ if h.mode != modeRollout || h.ro == nil {
+ continue
+ }
+ if _, ok := boxes[h.ro.Box]; !ok {
+ labels = append(labels, h.ro.Box)
+ }
+ boxes[h.ro.Box] = append(boxes[h.ro.Box], h)
+ }
+ sort.Strings(labels)
+ return boxes, labels
+}
+
+// boxDone is spec §9.1's rule: every targeted instance on the box is final,
+// and its scheduler has nothing queued, running or settling for the app, by
+// the latest word the watch has from the box.
+func boxDone(hs []*watchedHost) bool {
+ for _, h := range hs {
+ if !h.state.final() || !h.ro.SchedulerIdle {
+ return false
+ }
+ }
+ return true
+}
+
+func boxSummary(label string, hs []*watchedHost) string {
+ counts := map[rolloutState]int{}
+ for _, h := range hs {
+ counts[h.state]++
+ }
+ var parts []string
+ for _, p := range []struct {
+ s rolloutState
+ format string
+ }{{rolloutSwapped, watchCountSwapped}, {rolloutAsleep, watchCountAsleep}, {rolloutFailed, watchCountFailed}, {rolloutHeld, watchCountHeld}} {
+ if counts[p.s] > 0 {
+ parts = append(parts, fmt.Sprintf(p.format, counts[p.s]))
+ }
+ }
+ return fmt.Sprintf(watchBox, label, strings.Join(parts, ", "))
+}
+
+// stayedOn is the build a held instance is still on: what it serves, or,
+// asleep, its last good build.
+func stayedOn(r *apiRollout) string {
+ if r.Serving != "" {
+ return r.Serving
+ }
+ return r.Good
+}
+
+func (rw rolloutWatch) printHost(c colors, h *watchedHost) {
+ host := h.t.Host
+ if h.mode == modeURL {
+ fmt.Fprintf(rw.w, "%s https://%s\n", c.ok("live"), host)
+ return
+ }
+ switch h.state {
+ case rolloutSwapped:
+ fmt.Fprintf(rw.w, "%s https://%s\n", c.ok(watchWordSwapped), host)
+ case rolloutAsleep:
+ fmt.Fprintf(rw.w, "%s https://%s (%s)\n", c.ok(watchWordInstalled), host, watchAsleepTail)
+ case rolloutFailed:
+ fmt.Fprintf(rw.w, "%s https://%s: %s\n", c.warn(watchWordFailed), host, rolloutwords.Reason(h.fin.Reason))
+ case rolloutHeld:
+ fmt.Fprintf(rw.w, "%s https://%s: %s\n", c.warn(watchWordHeld), host, fmt.Sprintf(watchHeldTail, stayedOn(h.fin), rw.ver))
+ }
+}
+
+func (rw rolloutWatch) result(hosts []*watchedHost) error {
+ failed, held, down := 0, 0, false
+ for _, h := range hosts {
+ if h.mode != modeRollout {
+ continue
+ }
+ switch h.state {
+ case rolloutFailed:
+ failed++
+ down = down || h.fin.Attempt == status.AttemptDown
+ case rolloutHeld:
+ held++
+ }
+ }
+ if failed == 0 && held == 0 {
+ return nil
+ }
+ msg := fmt.Sprintf(watchFailedSummary, rw.ver, failed, held, rw.app, rw.ver, rw.channel)
+ if down {
+ msg += "\n" + fmt.Sprintf(watchHoldDownSummary, rw.app)
+ }
+ return errors.New(msg)
+}
+
+// pendingDetail is what a timeout says about a host still pending.
+func (rw rolloutWatch) pendingDetail(h *watchedHost) string {
+ r := h.ro
+ switch {
+ case h.mode == modeUnknown:
+ return watchPendingUnlisted
+ case h.mode == modeURL:
+ return h.urlDiag
+ case h.dark:
+ return watchPendingDark
+ case r.Unavailable != "":
+ return rolloutwords.Unavailable(r.Unavailable)
+ case r.Installed == "":
+ return watchPendingNotInstalled
+ case r.Installed != rw.ver:
+ return fmt.Sprintf(watchPendingOther, r.Installed)
+ case r.Attempt == status.AttemptQueued:
+ return watchPendingQueued
+ case r.Attempt == status.AttemptRunning:
+ return watchPendingRunning
+ case r.Attempt == status.AttemptSettling:
+ return watchPendingSettling
+ }
+ return watchPendingWaiting
+}
+
+func (rw rolloutWatch) timedOut(hosts []*watchedHost, boxes map[string][]*watchedHost, labels []string) error {
+ var b strings.Builder
+ fmt.Fprintf(&b, watchTimeout, rw.ver)
+ var pending []*watchedHost
+ for _, h := range hosts {
+ if !h.done() {
+ pending = append(pending, h)
+ }
+ }
+ sort.Slice(pending, func(i, j int) bool { return pending[i].t.Host < pending[j].t.Host })
+ for _, h := range pending {
+ fmt.Fprintf(&b, "\n %s: %s", h.t.Host, rw.pendingDetail(h))
+ // The reason a wake failed, from the platform log stream, as
+ // watchVersion gives it: only for a host polled by URL, whose poll
+ // is what woke it.
+ if h.mode == modeURL && rw.diag != nil && rw.diag.wakeReason != nil {
+ if r := rw.diag.wakeReason(h.t.Host); r != "" {
+ fmt.Fprintf(&b, "\n %s", r)
+ }
+ }
+ }
+ for _, l := range labels {
+ allFinal := true
+ for _, h := range boxes[l] {
+ allFinal = allFinal && h.state.final()
+ }
+ if allFinal && !boxDone(boxes[l]) {
+ fmt.Fprintf(&b, "\n "+watchBoxBusy, l)
+ }
+ }
+ if rw.diag != nil && rw.diag.logsHint != "" {
+ fmt.Fprintf(&b, "\nfull timeline: %s", rw.diag.logsHint)
+ }
+ return errors.New(b.String())
+}
+
+// watchDeployed is the watch for an instance (non-static) deploy of ver onto
+// channel.
+func watchDeployed(ctx context.Context, sess *apiSession, acct apiAcct, app, ver, channel string, hosts []string, probe string, timeout time.Duration) error {
+ targets := make([]watchTarget, 0, len(hosts))
+ for _, h := range hosts {
+ targets = append(targets, watchTarget{Host: h, Origin: watchOrigin(h), Probe: probe})
+ }
+ return newDeployWatch(ctx, sess, acct, app, ver, channel, probe, timeout).run(targets)
+}
+
+// newDeployWatch is watchDeployed's watch, built apart so a test can read
+// its cadence without a console.
+func newDeployWatch(ctx context.Context, sess *apiSession, acct apiAcct, app, ver, channel, probe string, timeout time.Duration) rolloutWatch {
+ client := &http.Client{Timeout: 10 * time.Second}
+ return rolloutWatch{
+ w: stdout, app: app, ver: ver, channel: channel, timeout: timeout,
+ interval: deployWatchPollEvery, listEvery: deployWatchListEvery,
+ list: func() ([]apiInstanceRecord, bool, error) {
+ recs, truncated, _, err := sess.instanceList(ctx, acct, app, "", "")
+ return recs, truncated, err
+ },
+ check: func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, ver, n) },
+ onURL: func() { printWatchNote(probe, false, app) },
+ diag: deployDiagnostics(ctx, sess, acct, app),
+ }
+}
+
+// deployDiagnostics is the timeout report's help, moved out of runDeploy
+// unchanged.
+func deployDiagnostics(ctx context.Context, sess *apiSession, acct apiAcct, app string) *watchDiagnostics {
+ return &watchDiagnostics{
+ // A timed-out host's wake reason, from the platform log stream —
+ // scoped tight (one host, errors only, the watch's own recent past)
+ // so a diagnosis this may print was caused by THIS deploy, not an
+ // incident from earlier in the day.
+ wakeReason: func(host string) string {
+ q := url.Values{"stream": {"platform"}, "level": {"error"},
+ "host": {host}, "since": {"15m"}, "limit": {"1"}}
+ res, err := sess.logsGet(ctx, acct, app, q)
+ if err != nil || len(res.Records) == 0 {
+ return ""
+ }
+ return res.Records[len(res.Records)-1].Msg
+ },
+ logsHint: fmt.Sprintf("carlos logs --app %s --account %s --stream platform", app, acct.Seg),
+ }
+}
+
+// printWatchNote is runDeploy's probe-or-header note, moved out unchanged.
+func printWatchNote(probe string, staticDeploy bool, app string) {
+ if probe != "" {
+ mutedf(stdout, "watching %s for the build %s reports itself\n", probe, app)
+ return
+ }
+ // Said out loud, every time, because the silence is what the bug was
+ // made of (discussion #13). With no probe this watch can only ask
+ // the ROUTER what release was adopted; it cannot ask the process
+ // what it is running. A deploy that goes green here has proved that
+ // adoption happened, not that anything restarted — and an app whose
+ // instance failed to cycle looks identical. It is a one-line fix
+ // per app, so the line names it rather than describing the problem
+ // and leaving the reader to find the verb.
+ mutedf(stdout, "watching the X-Carlos-Version header: it reports the release ADOPTED, not the build running\n")
+ // The advisory below assumes a process to probe. A static route has
+ // none — no /api/version, no exec child to cycle — so the suggested
+ // command would only teach the operator to configure a probe that
+ // 404s, which then fails and times out every future `carlos deploy`
+ // for that site. Static deploys get the line above (still true: the
+ // header reports adoption, not a running build) and stop there.
+ if !staticDeploy {
+ mutedf(stdout, " a probe proves the process actually cycled — carlos apps build-probe --app %s /api/version\n", app)
+ }
+}
diff --git a/deploywatch_test.go b/deploywatch_test.go
new file mode 100644
index 000000000..e76d47fef
--- /dev/null
+++ b/deploywatch_test.go
@@ -0,0 +1,555 @@
+package main
+
+import (
+ "bytes"
+ "errors"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/rolloutwords"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+const B = "b2"
+
+func TestClassifyRollout(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ r apiRollout
+ want rolloutState
+ }{
+ {"box has not installed B", apiRollout{Installed: "b1", Serving: "b1", Awake: true}, rolloutPending},
+ {"idle scheduler, B not installed", apiRollout{Installed: "b1", Good: "b1", SchedulerIdle: true}, rolloutPending},
+ {"swapped", apiRollout{Installed: B, Serving: B, Awake: true}, rolloutSwapped},
+ {"asleep with B installed", apiRollout{Installed: B, Good: "b1"}, rolloutAsleep},
+ {"queued", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
+ {"canary settling", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
+ {"failed here", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth}, rolloutFailed},
+ {"hold mode, down", apiRollout{Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B}, rolloutFailed},
+ {"held by a sibling's failure", apiRollout{Installed: B, Serving: "b1", Awake: true, HeldBy: B}, rolloutHeld},
+ {"asleep and held", apiRollout{Installed: B, Good: "b1", HeldBy: B}, rolloutHeld},
+ {"an older build's failure is not B's", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptFailed, AttemptBuild: "b0"}, rolloutSwapped},
+ // R8: an open attempt is pending whatever awake says, and B serving
+ // is not swapped until the canary's settle closes.
+ {"R8 canary serving B, settle open", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
+ {"R8 first wake in flight", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
+ {"R8 asleep, queued", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
+ {"R8 asleep, running", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptRunning, AttemptBuild: B}, rolloutPending},
+ // R6: a box that cannot arm reports no per-host state: its awake
+ // false is meaningless, so B installed there is not asleep.
+ {"R6 refusing box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}, rolloutPending},
+ {"R6 declined box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "generations-off"}, rolloutPending},
+ } {
+ r := tc.r
+ if got := classifyRollout(&r, B); got != tc.want {
+ t.Errorf("%s: %s, want %s", tc.name, got, tc.want)
+ }
+ }
+}
+
+// sweep is one scripted answer of the instance list.
+type sweep struct {
+ recs []apiInstanceRecord
+ truncated bool
+ err error
+}
+
+// scripted answers each list call with the next sweep, repeating the last.
+func scripted(calls *atomic.Int64, steps ...sweep) func() ([]apiInstanceRecord, bool, error) {
+ return func() ([]apiInstanceRecord, bool, error) {
+ i := int(calls.Add(1)) - 1
+ s := steps[min(i, len(steps)-1)]
+ return s.recs, s.truncated, s.err
+ }
+}
+
+// sweeps answers each list call with the next set, repeating the last; no
+// list is truncated.
+func sweeps(calls *atomic.Int64, sets ...[]apiInstanceRecord) func() ([]apiInstanceRecord, bool, error) {
+ steps := make([]sweep, 0, len(sets))
+ for _, s := range sets {
+ steps = append(steps, sweep{recs: s})
+ }
+ return scripted(calls, steps...)
+}
+
+func rec(host string, r *apiRollout) apiInstanceRecord {
+ return apiInstanceRecord{Host: host, Rollout: r}
+}
+
+func targetsFor(hosts ...string) []watchTarget {
+ var out []watchTarget
+ for _, h := range hosts {
+ out = append(out, watchTarget{Host: h, Origin: "http://127.0.0.1:1"})
+ }
+ return out
+}
+
+// noURL fails the test if a host is polled over HTTP: a reported host must
+// never be requested, because a request wakes an asleep instance.
+func noURL(t *testing.T) func(watchTarget, int64) (bool, string) {
+ var told sync.Map
+ return func(tg watchTarget, _ int64) (bool, string) {
+ if _, dup := told.LoadOrStore(tg.Host, true); !dup {
+ t.Errorf("%s was polled over HTTP", tg.Host)
+ }
+ return false, ""
+ }
+}
+
+// pollLog records which hosts were polled over HTTP and answers live as told.
+type pollLog struct {
+ mu sync.Mutex
+ polled []string
+}
+
+func (p *pollLog) check(live map[string]bool) func(watchTarget, int64) (bool, string) {
+ return func(tg watchTarget, _ int64) (bool, string) {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ p.polled = append(p.polled, tg.Host)
+ if live[tg.Host] {
+ return true, ""
+ }
+ return false, "still serving b1"
+ }
+}
+
+func (p *pollLog) hosts() map[string]int {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ out := map[string]int{}
+ for _, h := range p.polled {
+ out[h]++
+ }
+ return out
+}
+
+func testWatch(w io.Writer, list func() ([]apiInstanceRecord, bool, error), check func(watchTarget, int64) (bool, string)) rolloutWatch {
+ return rolloutWatch{w: w, app: "shop", ver: B, channel: "edge", timeout: 300 * time.Millisecond,
+ interval: 5 * time.Millisecond, list: list, check: check}
+}
+
+func TestWatchRolloutFollowsTheBoxToDone(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ list := sweeps(&calls,
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: "b1", Good: "b1", SchedulerIdle: true})},
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
+ )
+ if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ for _, want := range []string{
+ fmt.Sprintf(watchIntro, B),
+ watchWordSwapped + " https://a.test",
+ watchWordInstalled + " https://b.test (" + watchAsleepTail + ")",
+ fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountSwapped, 1)+", "+fmt.Sprintf(watchCountAsleep, 1)),
+ } {
+ if !strings.Contains(out.String(), want) {
+ t.Errorf("output lacks %q:\n%s", want, out.String())
+ }
+ }
+ if calls.Load() < 3 {
+ t.Errorf("done after %d sweeps: an open settle must hold the box", calls.Load())
+ }
+}
+
+// R8: a canary already serving B is not swapped while its settle is open.
+func TestWatchRolloutCanaryServingBWaitsForItsSettle(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ list := sweeps(&calls,
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B, SchedulerIdle: false})},
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonSettle, SchedulerIdle: true})},
+ )
+ err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test"))
+ if err == nil || !strings.HasPrefix(err.Error(), fmt.Sprintf(watchFailedSummary, B, 1, 0, "shop", B, "edge")) {
+ t.Fatalf("err = %v\n%s", err, out.String())
+ }
+ if strings.Contains(out.String(), watchWordSwapped) {
+ t.Fatalf("a canary in its settle was counted swapped:\n%s", out.String())
+ }
+}
+
+// A host that is final keeps being refreshed: here the sleeper is final
+// (installed, asleep) from the first sweep, while its box's scheduler is
+// still busy, and the box goes idle a sweep later. A loop that skips final
+// hosts keeps the first report's SchedulerIdle=false for ever, so boxDone
+// never holds and a finished deploy times out. The host prints once.
+func TestWatchRolloutRefreshesAFinalHostUntilTheBoxIsDone(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ list := sweeps(&calls,
+ []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
+ []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
+ []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
+ )
+ if err := testWatch(&out, list, noURL(t)).run(targetsFor("b.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ line := watchWordInstalled + " https://b.test (" + watchAsleepTail + ")"
+ if got := strings.Count(out.String(), line); got != 1 {
+ t.Fatalf("%q printed %d times, want once:\n%s", line, got, out.String())
+ }
+ if calls.Load() < 3 {
+ t.Fatalf("done after %d sweeps, before the box's scheduler went idle", calls.Load())
+ }
+}
+
+// Review Focus 2.
+func TestWatchRolloutNoOpDeployEndsAtOnce(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: B, SchedulerIdle: true}),
+ })
+ if err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
+ t.Fatal(err)
+ }
+ if calls.Load() != 1 {
+ t.Fatalf("an already-converged deploy took %d sweeps, want 1", calls.Load())
+ }
+}
+
+func TestWatchRolloutIdleSchedulerBeforeInstallIsPending(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true})})
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
+ if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchTimeout, B)) || !strings.Contains(err.Error(), fmt.Sprintf(watchPendingOther, "b1")) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+// Review Focus 4: a newer build replaced the watched one.
+func TestWatchRolloutNewerBuildNamesWhatIsInstalled(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true})})
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
+ if err == nil || !strings.Contains(err.Error(), "a.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+func TestWatchRolloutOpenSettleHoldsTheWatch(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"}),
+ })
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
+ if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchBoxBusy, "x1")) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+// R7: the host the build failed on is failed; a sibling its record holds
+// back is held.
+func TestWatchRolloutReportsFailureAndNamesPromoteRetry(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
+ rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
+ })
+ err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "c.test"))
+ if err == nil || err.Error() != fmt.Sprintf(watchFailedSummary, B, 1, 2, "shop", B, "edge") {
+ t.Fatalf("err = %v", err)
+ }
+ for _, want := range []string{
+ watchWordFailed + " https://a.test: " + rolloutwords.Reason(status.ReasonHealth),
+ watchWordHeld + " https://b.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
+ watchWordHeld + " https://c.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
+ fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountFailed, 1)+", "+fmt.Sprintf(watchCountHeld, 2)),
+ } {
+ if !strings.Contains(out.String(), want) {
+ t.Errorf("output lacks %q:\n%s", want, out.String())
+ }
+ }
+ if strings.Contains(err.Error(), watchHoldDownSummary[:20]) {
+ t.Errorf("an overlap failure named the hold-down rollback: %v", err)
+ }
+}
+
+// R7: a hold failure (attempt down for B) is failed, and the summary adds
+// the hold-down sentence.
+func TestWatchRolloutHoldDownNamesRollback(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B, Reason: status.ReasonExit, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
+ })
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
+ want := fmt.Sprintf(watchFailedSummary, B, 1, 1, "shop", B, "edge") + "\n" + fmt.Sprintf(watchHoldDownSummary, "shop")
+ if err == nil || err.Error() != want {
+ t.Fatalf("err = %v\nwant %s", err, want)
+ }
+}
+
+// Review Focus 3 and R1: a box that stops reporting never falls back to URL
+// polling.
+func TestWatchRolloutDarkBoxNeverPollsOverHTTP(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls,
+ []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true})},
+ []apiInstanceRecord{rec("a.test", nil)},
+ )
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
+ if err == nil || !strings.Contains(err.Error(), "a.test: "+watchPendingDark) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+// R2: a final state is final for the life of the watch. a.test swaps and
+// its box goes dark; d.test swaps and its box then reports a newer build.
+// Neither returns to pending, so the watch ends once x2's canary settles.
+func TestWatchRolloutFinalIsStickyThroughADarkBoxAndANewerBuild(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ list := sweeps(&calls,
+ []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("d.test", &apiRollout{Box: "x3", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
+ },
+ []apiInstanceRecord{
+ rec("a.test", nil),
+ rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
+ },
+ []apiInstanceRecord{
+ rec("a.test", nil),
+ rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ },
+ )
+ if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "d.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ for _, h := range []string{"a.test", "b.test", "d.test"} {
+ if got := strings.Count(out.String(), watchWordSwapped+" https://"+h+"\n"); got != 1 {
+ t.Errorf("%s printed swapped %d times:\n%s", h, got, out.String())
+ }
+ }
+}
+
+// R3: absence from a truncated list keeps a host where it was. b.test is
+// missing from the first (truncated) sweep and the list then fails once;
+// it is never polled over HTTP, and its report, when it comes, decides it.
+// a.test is missing from a later truncated sweep and keeps its state.
+func TestWatchRolloutTruncatedListNeverSwitchesAHostToURL(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ busyA := rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B})
+ list := scripted(&calls,
+ sweep{recs: []apiInstanceRecord{busyA}, truncated: true},
+ sweep{err: errors.New("instances list: 429")},
+ sweep{recs: []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true})}, truncated: true},
+ sweep{recs: []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true}),
+ }},
+ )
+ if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ if calls.Load() < 4 {
+ t.Fatalf("done after %d sweeps: a.test was still settling when the list cut it off", calls.Load())
+ }
+}
+
+// R3: a host no truncated list ever names stays unknown, and the timeout
+// says why.
+func TestWatchRolloutAlwaysTruncatedNamesTheUnlistedHost(t *testing.T) {
+ var calls atomic.Int64
+ list := scripted(&calls, sweep{recs: []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ }, truncated: true})
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
+ if err == nil || !strings.Contains(err.Error(), "b.test: "+watchPendingUnlisted) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+// R4: a host its box did not report on at first (created since the box's
+// last snapshot) is watched by URL until a report comes, then by the report.
+func TestWatchRolloutALateReportMovesAHostOffURL(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ var pl pollLog
+ list := sweeps(&calls,
+ []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("c.test", nil),
+ },
+ []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}),
+ },
+ )
+ if err := testWatch(&out, list, pl.check(nil)).run(targetsFor("a.test", "c.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 {
+ t.Errorf("polled %v, want c.test once (before its report), a.test never", got)
+ }
+ if !strings.Contains(out.String(), watchWordInstalled+" https://c.test ("+watchAsleepTail+")") {
+ t.Errorf("c.test not followed through its report:\n%s", out.String())
+ }
+}
+
+func TestWatchRolloutMixedFleetPollsOnlyTheUnreported(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ var pl pollLog
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("c.test", nil),
+ })
+ rw := testWatch(&out, list, pl.check(map[string]bool{"c.test": true}))
+ onURL := 0
+ rw.onURL = func() { onURL++ }
+ if err := rw.run(targetsFor("a.test", "c.test")); err != nil {
+ t.Fatal(err)
+ }
+ if got := pl.hosts(); len(got) != 1 || got["c.test"] != 1 || onURL != 1 || !strings.Contains(out.String(), "live https://c.test") {
+ t.Fatalf("polled %v, onURL %d, output:\n%s", got, onURL, out.String())
+ }
+}
+
+// R6: a box that refuses every managed start holds its instances pending,
+// never polled, and the timeout names why in the box's words.
+func TestWatchRolloutRefusingBoxIsPendingWithItsReason(t *testing.T) {
+ var calls atomic.Int64
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}),
+ })
+ err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
+ if err == nil || !strings.Contains(err.Error(), "a.test: "+rolloutwords.Unavailable("no-release-key")) {
+ t.Fatalf("err = %v", err)
+ }
+}
+
+// R6: a box that declined overlap runs today's stop-then-start paths, so its
+// instances are watched by URL, like a flag-off box's.
+func TestWatchRolloutDeclinedBoxIsWatchedByURL(t *testing.T) {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ var pl pollLog
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
+ rec("c.test", &apiRollout{Box: "x2", Installed: B, Unavailable: "generations-off"}),
+ })
+ if err := testWatch(&out, list, pl.check(map[string]bool{"c.test": true})).run(targetsFor("a.test", "c.test")); err != nil {
+ t.Fatalf("watch: %v\n%s", err, out.String())
+ }
+ if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 || !strings.Contains(out.String(), "live https://c.test") {
+ t.Fatalf("polled %v, output:\n%s", got, out.String())
+ }
+}
+
+// R5: the instance list is read on its own cadence (5 s in production), not
+// on every URL poll.
+func TestWatchRolloutListsOnItsOwnCadence(t *testing.T) {
+ var calls atomic.Int64
+ var pl pollLog
+ list := sweeps(&calls, []apiInstanceRecord{
+ rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
+ rec("c.test", nil),
+ })
+ rw := testWatch(io.Discard, list, pl.check(nil))
+ rw.timeout, rw.listEvery = 80*time.Millisecond, time.Hour
+ if err := rw.run(targetsFor("a.test", "c.test")); err == nil {
+ t.Fatal("watch ended, want a timeout")
+ }
+ if calls.Load() != 1 || pl.hosts()["c.test"] < 3 {
+ t.Fatalf("%d lists for %d URL polls: the list must keep its own cadence", calls.Load(), pl.hosts()["c.test"])
+ }
+}
+
+func TestDeployWatchCadence(t *testing.T) {
+ rw := newDeployWatch(nil, nil, apiAcct{Seg: "acme"}, "shop", B, "edge", "", time.Minute)
+ if rw.listEvery != 5*time.Second || rw.interval != 2*time.Second {
+ t.Fatalf("list every %s, poll every %s: want 5s and 2s", rw.listEvery, rw.interval)
+ }
+}
+
+// R10: with no box reporting, the watch is today's watchVersion, byte for
+// byte, its timeout report included.
+func TestWatchRolloutWithNoReportsIsTheOldWatch(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("X-Carlos-Version", B)
+ }))
+ defer srv.Close()
+ var out bytes.Buffer
+ var calls atomic.Int64
+ onURL := 0
+ rw := testWatch(&out, sweeps(&calls, []apiInstanceRecord{rec("x.test", nil)}), nil)
+ rw.onURL = func() { onURL++ }
+ if err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}}); err != nil {
+ t.Fatal(err)
+ }
+ if onURL != 1 || !strings.Contains(out.String(), "live https://x.test") || strings.Contains(out.String(), watchWordSwapped) {
+ t.Fatalf("onURL %d, output:\n%s", onURL, out.String())
+ }
+}
+
+func TestWatchRolloutWithNoReportsTimesOutAsTheOldWatch(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("X-Carlos-Version", "b1")
+ }))
+ defer srv.Close()
+ targets := []watchTarget{{Host: "x.test", Origin: srv.URL}, {Host: "y.test", Origin: srv.URL}}
+ diag := &watchDiagnostics{logsHint: "carlos logs --app shop"}
+ var oldOut bytes.Buffer
+ oldErr := watchVersion(&oldOut, targets, B, 30*time.Millisecond, 5*time.Millisecond, diag)
+ for name, recs := range map[string][]apiInstanceRecord{
+ "flag off": {rec("x.test", nil), rec("y.test", nil)},
+ // R6: every box declined overlap.
+ "declined": {rec("x.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-cgroup-delegation"}), rec("y.test", nil)},
+ // A report on an instance the deploy does not target is not one.
+ "untargeted report": {rec("x.test", nil), rec("y.test", nil), rec("z.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true})},
+ } {
+ var out bytes.Buffer
+ var calls atomic.Int64
+ rw := testWatch(&out, sweeps(&calls, recs), nil)
+ rw.timeout, rw.diag = 30*time.Millisecond, diag
+ err := rw.run(targets)
+ if err == nil || oldErr == nil || err.Error() != oldErr.Error() || out.String() != oldOut.String() {
+ t.Errorf("%s: err = %v\nwant %v\nout %q want %q", name, err, oldErr, out.String(), oldOut.String())
+ }
+ }
+}
+
+// R9: the until-about form only for a held_until in the future; otherwise
+// today's line, byte for byte.
+func TestDeployHeldLine(t *testing.T) {
+ now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
+ today := "shop is a production app: the stable bake window applies — the fleet adopts when it expires; console fast-track if this is an emergency"
+ future := now.Add(3 * time.Hour)
+ for _, tc := range []struct {
+ name, until, want string
+ }{
+ {"absent", "", today},
+ {"past", now.Add(-time.Minute).Format(time.RFC3339), today},
+ {"now", now.Format(time.RFC3339), today},
+ {"unparseable", "soon", today},
+ {"future", future.Format(time.RFC3339), fmt.Sprintf(deployHeldUntil, "shop", "stable", future.Local().Format("Mon 2 Jan 15:04 MST"))},
+ } {
+ if got := deployHeldLine("shop", "stable", tc.until, now); got != tc.want {
+ t.Errorf("%s: %q\nwant %q", tc.name, got, tc.want)
+ }
+ }
+}