amadan
zdd-6 final fix wave 1: W2 the deploy watch falls back only on evidence, inside one deadline
Failed lists (5xx, 429, network) are retried until the deadline and the timeout names the list failure (zdd.watch.list-failed); the watch falls back to watchVersion only on a 4xx other than 429, or on a list where every target is listed, the rollout join was complete and no target is followed; the fallback gets the remaining time. The console's instance list adds rollout_incomplete (omitted when false) from status.LoadBoxesComplete. Astra plan 6 R1 #2, #4, #7.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
5 files changed,
+407
−74
deploywatch.go+142 −48deploywatch_test.go+196 −14internal/console/apiinstances.go+28 −11internal/console/rolloutjoin_test.go+40 −0internal/provision/plan9_b8_census_test.go+1 −1
diff --git a/deploywatch.go b/deploywatch.go| index 87cc84c4c..b97c1c412 100644 |
| --- a/deploywatch.go |
| +++ b/deploywatch.go |
| @@ -10,6 +10,7 @@ package main |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| "io" |
| @@ -73,17 +74,70 @@ const ( |
| 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" |
| + // TODO-COPY zdd.watch.list-failed |
| + watchListFailed = "the console's instance list could not be read, so the rollout could not be followed" |
| + // TODO-COPY zdd.watch.pending-unread (new: no plan draft; fix wave 1, W2) |
| + watchPendingUnread = "the console could not read every box's report, so its state is unknown" |
| ) |
| +// watchList is one answer of the console's instance list, as the watch |
| +// reads it. incomplete is the console's rollout_incomplete: some box's |
| +// report could not be read, so a host with no report may yet have one. |
| +type watchList struct { |
| + recs []apiInstanceRecord |
| + truncated bool |
| + incomplete bool |
| +} |
| + |
| +// listStatusError is a list the console answered with a status other than |
| +// 200. The status decides what the watch makes of it (listAbsent). |
| +type listStatusError struct { |
| + status int |
| + err error |
| +} |
| + |
| +func (e *listStatusError) Error() string { return e.err.Error() } |
| +func (e *listStatusError) Unwrap() error { return e.err } |
| + |
| +// listAbsent reports whether a failed list is affirmative evidence that no |
| +// box reports on this app (Astra plan 6 R1 #4): a 4xx other than 429 says |
| +// the console has no list for it (a console from before the door, an app |
| +// with no instances). A 5xx, a 429, a network error or an unreadable answer |
| +// says nothing about the boxes. |
| +func listAbsent(err error) bool { |
| + var se *listStatusError |
| + return errors.As(err, &se) && se.status >= 400 && se.status < 500 && se.status != http.StatusTooManyRequests |
| +} |
| + |
| +// watchInstanceList is the watch's read of the instance list. It is |
| +// instanceList's request, kept apart because the watch needs what that |
| +// method drops: the status of a refusal, and rollout_incomplete. |
| +func (s *apiSession) watchInstanceList(ctx context.Context, a apiAcct, app string) (watchList, error) { |
| + status, body, err := s.getJSON(ctx, apiPath(a, app, "/instances")) |
| + if err != nil { |
| + return watchList{}, err |
| + } |
| + if status != http.StatusOK { |
| + return watchList{}, &listStatusError{status: status, err: apiErrorFromBody(status, body)} |
| + } |
| + var res struct { |
| + Instances []apiInstanceRecord `json:"instances"` |
| + Truncated bool `json:"truncated"` |
| + RolloutIncomplete bool `json:"rollout_incomplete"` |
| + } |
| + if err := json.Unmarshal(body, &res); err != nil { |
| + return watchList{}, fmt.Errorf("instances list: unreadable response: %w", err) |
| + } |
| + return watchList{recs: res.Instances, truncated: res.Truncated, incomplete: res.RolloutIncomplete}, nil |
| +} |
| + |
| // 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. |
| -// firstListTries is how many times a failing first list is asked before the |
| -// watch falls back to watchVersion. |
| +// watched by URL is still polled every deployWatchPollEvery, as before. A |
| +// failing first list is asked again on the same cadence until the deadline. |
| const ( |
| - firstListTries = 3 |
| deployWatchListEvery = 5 * time.Second |
| deployWatchPollEvery = 2 * time.Second |
| ) |
| @@ -164,7 +218,7 @@ func follows(r *apiRollout) bool { |
| type watchMode int |
| const ( |
| - modeUnknown watchMode = iota // not yet listed (a truncated list left it out) |
| + modeUnknown watchMode = iota // not yet decided: a truncated list left it out, or an incomplete join had no report for it |
| modeRollout // watched through its box's report, for the rest of the watch |
| modeURL // watched by URL, as watchVersion does, until a report comes |
| ) |
| @@ -177,8 +231,11 @@ type watchedHost struct { |
| 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 |
| + dark bool |
| + // unread is set while the host is undecided because the console's |
| + // rollout join was incomplete, not because the list was cut short. |
| + unread bool |
| + state rolloutState |
| // fin is the report the host went final on. A final state survives |
| // missing observations, and later reports that agree the build is |
| // installed: what it printed and what it counts come from fin. A report |
| @@ -199,9 +256,9 @@ func (h *watchedHost) done() bool { |
| 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 |
| +// rolloutWatch is one watch. list is the console's instance list, asked to |
| +// answer by the watch's deadline; 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. |
| @@ -210,34 +267,42 @@ type rolloutWatch struct { |
| app, ver, channel string |
| timeout, interval time.Duration |
| listEvery time.Duration |
| - list func() ([]apiInstanceRecord, bool, error) |
| + list func(deadline time.Time) (watchList, error) |
| check func(t watchTarget, n int64) (bool, string) |
| onURL func() |
| diag *watchDiagnostics |
| } |
| func (rw rolloutWatch) run(targets []watchTarget) error { |
| + // One deadline governs the whole watch (Astra plan 6 R1 #2): every list |
| + // request, every wait between lists, and what is left for watchVersion. |
| deadline := time.Now().Add(rw.timeout) |
| - // A failed first list (a rate limit, a console restart) says nothing |
| - // about the boxes, so it is retried on the list's cadence before the |
| - // watch gives up on reports (fix round 1). |
| - recs, truncated, err := rw.list() |
| - for tries := 1; err != nil && tries < firstListTries; tries++ { |
| - time.Sleep(max(rw.listEvery, rw.interval)) |
| - recs, truncated, err = rw.list() |
| + // A failed list (a 5xx, a rate limit, a console restart, the network) |
| + // says nothing about the boxes, so it is asked again on the list's |
| + // cadence until the deadline, and never read as "no reports" (Astra plan |
| + // 6 R1 #4: a URL poll would wake an asleep instance on a reporting box). |
| + l, err := rw.list(deadline) |
| + for err != nil && !listAbsent(err) { |
| + left := time.Until(deadline) |
| + if left <= 0 { |
| + return rw.listFailed(err) |
| + } |
| + time.Sleep(min(max(rw.listEvery, rw.interval), left)) |
| + if time.Until(deadline) <= 0 { |
| + return rw.listFailed(err) |
| + } |
| + l, err = rw.list(deadline) |
| } |
| lastList := time.Now() |
| - // Today's watch, byte for byte (R10), only when the console has shown |
| - // the whole list and no box reports a rollout for any target; or when |
| - // it never answered, so a broken list endpoint still leaves the deploy |
| - // watched as it always was. A truncated list never falls back: a host |
| - // it cut off may sit on a reporting box, and a URL poll would wake it |
| - // (R3); its listed, unreported hosts are polled by URL anyway (R4). |
| - if err != nil || (!truncated && !reportsAny(recs, targets)) { |
| + // Today's watch, byte for byte (R10), only on affirmative evidence that |
| + // no target is on a reporting box: the console has no list for the app |
| + // (listAbsent), or every target is listed, the console read every box's |
| + // report, and none of the targets has one the watch follows (legacyOnly). |
| + if err != nil || legacyOnly(l, targets) { |
| if rw.onURL != nil { |
| rw.onURL() |
| } |
| - return watchVersion(rw.w, targets, rw.ver, rw.timeout, rw.interval, rw.diag) |
| + return watchVersion(rw.w, targets, rw.ver, time.Until(deadline), rw.interval, rw.diag) |
| } |
| var hosts []*watchedHost |
| seen := map[string]bool{} |
| @@ -249,7 +314,7 @@ func (rw rolloutWatch) run(targets []watchTarget) error { |
| } |
| c := colorFor(rw.w) |
| mutedf(rw.w, watchIntro, rw.ver) |
| - rw.apply(hosts, recs, truncated) |
| + rw.apply(hosts, l) |
| boxPrinted := map[string]bool{} |
| announcedURL := false |
| var n int64 |
| @@ -309,13 +374,13 @@ func (rw rolloutWatch) run(targets []watchTarget) error { |
| } |
| time.Sleep(rw.wait(hosts, lastList, deadline)) |
| if time.Since(lastList) >= rw.listEvery { |
| - recs, truncated, err = rw.list() |
| + l, err = rw.list(deadline) |
| 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) |
| + rw.apply(hosts, l) |
| } |
| } |
| } |
| @@ -339,20 +404,39 @@ func (rw rolloutWatch) wait(hosts []*watchedHost, lastList, deadline time.Time) |
| 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 |
| +// legacyOnly is the evidence the fallback to watchVersion needs (Astra plan |
| +// 6 R1 #4, #7): every target is on the list, the console's rollout join read |
| +// every box, and no target has a report the watch follows. It is decided on |
| +// the targets alone. A list cut short for the whole app neither causes nor |
| +// blocks it, and 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 |
| +func legacyOnly(l watchList, targets []watchTarget) bool { |
| + if l.incomplete { |
| + return false |
| + } |
| + listed := make(map[string]*apiRollout, len(l.recs)) |
| + for _, r := range l.recs { |
| + listed[r.Host] = r.Rollout |
| } |
| - for _, r := range recs { |
| - if want[r.Host] && follows(r.Rollout) { |
| - return true |
| + for _, t := range targets { |
| + ro, ok := listed[t.Host] |
| + if !ok || follows(ro) { |
| + return false |
| } |
| } |
| - return false |
| + return true |
| +} |
| + |
| +// listFailed is the timeout when no list answered before the deadline: |
| +// nothing was followed and nothing was polled, so it names the list. |
| +func (rw rolloutWatch) listFailed(err error) error { |
| + var b strings.Builder |
| + fmt.Fprintf(&b, watchTimeout, rw.ver) |
| + fmt.Fprintf(&b, "\n %s: %v", watchListFailed, err) |
| + if rw.diag != nil && rw.diag.logsHint != "" { |
| + fmt.Fprintf(&b, "\nfull timeline: %s", rw.diag.logsHint) |
| + } |
| + return errors.New(b.String()) |
| } |
| // apply folds one answered list into the hosts. |
| @@ -372,13 +456,17 @@ func reportsAny(recs []apiInstanceRecord, targets []watchTarget) bool { |
| // - 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 no followed report on a list whose rollout join was |
| +// incomplete keeps its mode and state too: its box's report may be the |
| +// one the console could not read (Astra plan 6 R1 #4). An undecided host |
| +// stays undecided, and is named so in a timeout. |
| // - 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 |
| +func (rw rolloutWatch) apply(hosts []*watchedHost, l watchList) { |
| + listed := make(map[string]*apiRollout, len(l.recs)) |
| + for i := range l.recs { |
| + listed[l.recs[i].Host] = l.recs[i].Rollout |
| } |
| for _, h := range hosts { |
| ro, ok := listed[h.t.Host] |
| @@ -390,7 +478,7 @@ func (rw rolloutWatch) apply(hosts []*watchedHost, recs []apiInstanceRecord, tru |
| // once it is final. |
| h.state, h.fin = rolloutPending, nil |
| } |
| - h.mode, h.ro, h.dark = modeRollout, ro, false |
| + h.mode, h.ro, h.dark, h.unread = modeRollout, ro, false, false |
| switch { |
| case !h.state.final(): |
| h.state = classifyRollout(ro, rw.ver) |
| @@ -402,9 +490,12 @@ func (rw rolloutWatch) apply(hosts []*watchedHost, recs []apiInstanceRecord, tru |
| case h.state != rolloutFailed && ro.Installed != rw.ver: |
| h.state, h.fin, h.printed = rolloutPending, nil, false |
| } |
| - case !ok && truncated: |
| + case !ok && l.truncated: |
| + h.unread = false |
| case h.mode == modeURL && h.urlLive: |
| // Live by URL is final too, until a report says otherwise. |
| + case l.incomplete: |
| + h.unread = h.mode == modeUnknown |
| case h.mode == modeRollout: |
| h.dark = true |
| default: |
| @@ -513,6 +604,8 @@ func (rw rolloutWatch) result(hosts []*watchedHost) error { |
| func (rw rolloutWatch) pendingDetail(h *watchedHost) string { |
| r := h.ro |
| switch { |
| + case h.mode == modeUnknown && h.unread: |
| + return watchPendingUnread |
| case h.mode == modeUnknown: |
| return watchPendingUnlisted |
| case h.mode == modeURL: |
| @@ -588,9 +681,10 @@ func newDeployWatch(ctx context.Context, sess *apiSession, acct apiAcct, app, ve |
| 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 |
| + list: func(deadline time.Time) (watchList, error) { |
| + lctx, cancel := context.WithDeadline(ctx, deadline) |
| + defer cancel() |
| + return sess.watchInstanceList(lctx, acct, app) |
| }, |
| check: func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, ver, n) }, |
| onURL: func() { printWatchNote(probe, false, app) }, |
diff --git a/deploywatch_test.go b/deploywatch_test.go| index 3d8a31282..e9e09ad44 100644 |
| --- a/deploywatch_test.go |
| +++ b/deploywatch_test.go |
| @@ -2,6 +2,7 @@ package main |
| import ( |
| "bytes" |
| + "context" |
| "errors" |
| "fmt" |
| "io" |
| @@ -56,23 +57,24 @@ func TestClassifyRollout(t *testing.T) { |
| // sweep is one scripted answer of the instance list. |
| type sweep struct { |
| - recs []apiInstanceRecord |
| - truncated bool |
| - err error |
| + recs []apiInstanceRecord |
| + truncated bool |
| + incomplete 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) { |
| +func scripted(calls *atomic.Int64, steps ...sweep) func(time.Time) (watchList, error) { |
| + return func(time.Time) (watchList, error) { |
| i := int(calls.Add(1)) - 1 |
| s := steps[min(i, len(steps)-1)] |
| - return s.recs, s.truncated, s.err |
| + return watchList{recs: s.recs, truncated: s.truncated, incomplete: s.incomplete}, 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) { |
| +func sweeps(calls *atomic.Int64, sets ...[]apiInstanceRecord) func(time.Time) (watchList, error) { |
| steps := make([]sweep, 0, len(sets)) |
| for _, s := range sets { |
| steps = append(steps, sweep{recs: s}) |
| @@ -132,7 +134,7 @@ func (p *pollLog) hosts() map[string]int { |
| return out |
| } |
| -func testWatch(w io.Writer, list func() ([]apiInstanceRecord, bool, error), check func(watchTarget, int64) (bool, string)) rolloutWatch { |
| +func testWatch(w io.Writer, list func(time.Time) (watchList, 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} |
| } |
| @@ -623,13 +625,15 @@ func TestWatchRolloutFirstListErrorIsRetried(t *testing.T) { |
| } |
| } |
| -// Fix round 1: three failed first lists fall back to today's watch, so a |
| -// broken list endpoint never leaves a fleet unwatched. |
| -func TestWatchRolloutThreeFailedFirstListsFallBack(t *testing.T) { |
| +// Astra plan 6 R1 #4 (reverses fix round 1's three-tries fallback): failed |
| +// lists are no evidence that the boxes do not report, so they are retried |
| +// until a list answers. Here the fourth does, with a report: the sleeper is |
| +// never requested. |
| +func TestWatchRolloutFailedFirstListsNeverFallBack(t *testing.T) { |
| origin, hits := hitServer(t) |
| var calls atomic.Int64 |
| var out bytes.Buffer |
| - failed := sweep{err: errors.New("instances list: 503")} |
| + failed := sweep{err: &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("instances list: 503")}} |
| list := scripted(&calls, failed, failed, failed, |
| sweep{recs: []apiInstanceRecord{rec("x.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})}}, |
| ) |
| @@ -637,13 +641,191 @@ func TestWatchRolloutThreeFailedFirstListsFallBack(t *testing.T) { |
| onURL := 0 |
| rw.onURL = func() { onURL++ } |
| if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil { |
| - t.Fatal(err) |
| + t.Fatalf("watch: %v\n%s", err, out.String()) |
| } |
| - if calls.Load() != 3 || onURL != 1 || hits.Load() == 0 || !strings.Contains(out.String(), "live https://x.test") { |
| + if calls.Load() != 4 || onURL != 0 || hits.Load() != 0 || !strings.Contains(out.String(), watchWordInstalled+" https://x.test") { |
| t.Fatalf("%d lists, onURL %d, %d HTTP polls, output:\n%s", calls.Load(), onURL, hits.Load(), out.String()) |
| } |
| } |
| +// Astra plan 6 R1 #4: a list that keeps failing with a 5xx, a 429 or a |
| +// network error is retried until the deadline. No origin is requested, and |
| +// the timeout names the list failure. |
| +func TestWatchRolloutListFailuresTimeOutNamingTheList(t *testing.T) { |
| + for name, lerr := range map[string]error{ |
| + "503": &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("could not list instances")}, |
| + "429": &listStatusError{status: http.StatusTooManyRequests, err: errors.New("too many requests")}, |
| + "network": errors.New("dial tcp: connection refused"), |
| + } { |
| + origin, hits := hitServer(t) |
| + var calls atomic.Int64 |
| + rw := testWatch(io.Discard, scripted(&calls, sweep{err: lerr}), nil) |
| + onURL := 0 |
| + rw.onURL = func() { onURL++ } |
| + err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}) |
| + if hits.Load() != 0 || onURL != 0 { |
| + t.Errorf("%s: %d HTTP polls, onURL %d: want none", name, hits.Load(), onURL) |
| + } |
| + want := fmt.Sprintf(watchTimeout, B) + "\n " + watchListFailed |
| + if err == nil || !strings.HasPrefix(err.Error(), want) || !strings.Contains(err.Error(), lerr.Error()) { |
| + t.Errorf("%s: err = %v\nwant it to start %q and carry the list's error", name, err, want) |
| + } |
| + if calls.Load() < 2 { |
| + t.Errorf("%s: %d lists: a failed list must be retried", name, calls.Load()) |
| + } |
| + } |
| +} |
| + |
| +// Astra plan 6 R1 #4: a 4xx other than 429 is a console with no list for |
| +// the app, so the watch is today's, as before. |
| +func TestWatchRolloutList404FallsBack(t *testing.T) { |
| + origin, hits := hitServer(t) |
| + var calls atomic.Int64 |
| + var out bytes.Buffer |
| + rw := testWatch(&out, scripted(&calls, sweep{err: &listStatusError{status: http.StatusNotFound, err: notFoundErr()}}), nil) |
| + onURL := 0 |
| + rw.onURL = func() { onURL++ } |
| + if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if calls.Load() != 1 || onURL != 1 || hits.Load() == 0 || out.String() != "live https://x.test\n" { |
| + t.Fatalf("%d lists, onURL %d, %d HTTP polls, output %q", calls.Load(), onURL, hits.Load(), out.String()) |
| + } |
| +} |
| + |
| +// Astra plan 6 R1 #4: a list whose rollout join was incomplete is no |
| +// evidence that a host has no report: it is not URL-polled, and the timeout |
| +// says why its state is unknown. |
| +func TestWatchRolloutIncompleteJoinIsNotNoReports(t *testing.T) { |
| + origin, hits := hitServer(t) |
| + var calls atomic.Int64 |
| + rw := testWatch(io.Discard, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil)}, incomplete: true}), nil) |
| + onURL := 0 |
| + rw.onURL = func() { onURL++ } |
| + err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}) |
| + if hits.Load() != 0 || onURL != 0 { |
| + t.Fatalf("%d HTTP polls, onURL %d: want none", hits.Load(), onURL) |
| + } |
| + if err == nil || !strings.Contains(err.Error(), "\n x.test: "+watchPendingUnread) { |
| + t.Fatalf("err = %v", err) |
| + } |
| +} |
| + |
| +// Astra plan 6 R1 #7: an app-wide truncated list on which every target is |
| +// listed with no report is the old watch, byte for byte, on success and on |
| +// timeout. |
| +func TestWatchRolloutTruncatedListWithEveryTargetUnreportedIsTheOldWatch(t *testing.T) { |
| + for _, served := range []string{B, "b1"} { |
| + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + w.Header().Set("X-Carlos-Version", served) |
| + })) |
| + 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) |
| + var out bytes.Buffer |
| + var calls atomic.Int64 |
| + // The production check, so a watch that wrongly stays on the reports |
| + // path prints what it would print, rather than panicking. |
| + client := &http.Client{Timeout: time.Second} |
| + rw := testWatch(&out, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil), rec("y.test", nil)}, truncated: true}), |
| + func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, B, n) }) |
| + rw.timeout, rw.diag = 30*time.Millisecond, diag |
| + err := rw.run(targets) |
| + srv.Close() |
| + if fmt.Sprint(err) != fmt.Sprint(oldErr) || out.String() != oldOut.String() { |
| + t.Errorf("serving %s: err = %v\nwant %v\nout %q\nwant %q", served, err, oldErr, out.String(), oldOut.String()) |
| + } |
| + } |
| +} |
| + |
| +// Astra plan 6 R1 #2: one deadline governs discovery. A list that keeps |
| +// failing at the production cadence (5 s between lists) does not hold a |
| +// 50 ms watch for ten seconds. |
| +func TestWatchRolloutTinyTimeoutHonouredThroughDiscovery(t *testing.T) { |
| + var calls atomic.Int64 |
| + rw := testWatch(io.Discard, scripted(&calls, sweep{err: &listStatusError{status: http.StatusBadGateway, err: errors.New("could not list instances")}}), nil) |
| + rw.timeout, rw.listEvery, rw.interval = 50*time.Millisecond, deployWatchListEvery, deployWatchPollEvery |
| + start := time.Now() |
| + err := rw.run(targetsFor("x.test")) |
| + if took := time.Since(start); took > time.Second { |
| + t.Fatalf("a 50ms watch took %s through a failing list", took) |
| + } |
| + if err == nil || !strings.Contains(err.Error(), watchListFailed) { |
| + t.Fatalf("err = %v", err) |
| + } |
| +} |
| + |
| +// Astra plan 6 R1 #2: the fallback gets what is left of the deadline, not |
| +// the whole timeout again. The list takes 300 ms to answer 404 inside a |
| +// 400 ms watch; today's watch then has about 100 ms, not 400. |
| +func TestWatchRolloutFallbackGetsTheRemainingTime(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() |
| + rw := testWatch(io.Discard, func(time.Time) (watchList, error) { |
| + time.Sleep(300 * time.Millisecond) |
| + return watchList{}, &listStatusError{status: http.StatusNotFound, err: notFoundErr()} |
| + }, nil) |
| + rw.timeout = 400 * time.Millisecond |
| + start := time.Now() |
| + err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}}) |
| + if took := time.Since(start); took > 600*time.Millisecond { |
| + t.Fatalf("a 400ms watch took %s: the fallback was given the whole timeout again", took) |
| + } |
| + if err == nil || !strings.Contains(err.Error(), "timed out waiting for x.test to serve "+B) { |
| + t.Fatalf("err = %v", err) |
| + } |
| +} |
| + |
| +// The production list reader, against a console: a 503 is retried to the |
| +// deadline with no origin request; a 404 with no JSON error falls back; and |
| +// rollout_incomplete is decoded, so an incomplete join is never read as no |
| +// reports. |
| +func TestDeployWatchReadsTheConsolesList(t *testing.T) { |
| + for _, tc := range []struct { |
| + name string |
| + answer func(w http.ResponseWriter) |
| + wantHits bool |
| + wantErr string |
| + }{ |
| + {"503", func(w http.ResponseWriter) { |
| + w.WriteHeader(http.StatusServiceUnavailable) |
| + io.WriteString(w, `{"error":"could not list instances"}`) |
| + }, false, watchListFailed + ": could not list instances"}, |
| + {"404", func(w http.ResponseWriter) { w.WriteHeader(http.StatusNotFound) }, true, ""}, |
| + {"incomplete join", func(w http.ResponseWriter) { |
| + io.WriteString(w, `{"instances":[{"host":"x.test"}],"truncated":false,"rollout_incomplete":true}`) |
| + }, false, "x.test: " + watchPendingUnread}, |
| + } { |
| + console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + if !strings.HasSuffix(r.URL.Path, "/instances") { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + tc.answer(w) |
| + })) |
| + origin, hits := hitServer(t) |
| + sess := &apiSession{c: newConsoleClient(console.URL, "tok"), console: console.URL} |
| + rw := newDeployWatch(context.Background(), sess, apiAcct{Seg: "acme"}, "shop", B, "edge", "", 300*time.Millisecond) |
| + rw.w, rw.onURL, rw.diag = io.Discard, nil, nil |
| + start := time.Now() |
| + err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}) |
| + took := time.Since(start) |
| + console.Close() |
| + if (hits.Load() > 0) != tc.wantHits { |
| + t.Errorf("%s: %d origin requests", tc.name, hits.Load()) |
| + } |
| + if tc.wantErr == "" && err != nil || tc.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tc.wantErr)) { |
| + t.Errorf("%s: err = %v, want %q", tc.name, err, tc.wantErr) |
| + } |
| + if took > 2*time.Second { |
| + t.Errorf("%s: a 300ms watch took %s", tc.name, took) |
| + } |
| + } |
| +} |
| + |
| // Fix round 1 (R3 on the first sweep): a truncated first list with no |
| // report among the listed targets is no evidence that no box reports. |
| // c.test (listed, unreported) is watched by URL; b.test (cut off, then |
diff --git a/internal/console/apiinstances.go b/internal/console/apiinstances.go| index 6a1a123d3..c27307fe6 100644 |
| --- a/internal/console/apiinstances.go |
| +++ b/internal/console/apiinstances.go |
| @@ -416,15 +416,26 @@ func (j boxJoin) dbSize(host string) *int64 { |
| // runs — but the nil check is kept here too so this helper degrades safely |
| // for any future caller that reaches it before that gate. |
| func loadBoxJoin(ctx context.Context, s release.Store, now time.Time) boxJoin { |
| + j, _ := loadBoxJoinComplete(ctx, s, now) |
| + return j |
| +} |
| + |
| +// loadBoxJoinComplete is loadBoxJoin that also says whether every box's |
| +// snapshot was read (status.LoadBoxesComplete): false when the listing |
| +// failed, was capped, or skipped an object it could not read. The join is |
| +// still best-effort; completeness is what lets a reader tell "no box reports |
| +// this host" from "the report may be the one that was not read" (Astra plan |
| +// 6 R1 #4: the deploy watch must not URL-poll on the second). |
| +func loadBoxJoinComplete(ctx context.Context, s release.Store, now time.Time) (boxJoin, bool) { |
| if s == nil { |
| - return boxJoin{} |
| + return boxJoin{}, false |
| } |
| - boxes, err := status.LoadBoxes(ctx, s) |
| + boxes, complete, err := status.LoadBoxesComplete(ctx, s) |
| if err != nil { |
| log.Printf("console: api instances: could not load status boxes for the list join: %v", err) |
| - return boxJoin{} |
| + return boxJoin{}, false |
| } |
| - return joinBoxes(boxes, now) |
| + return joinBoxes(boxes, now), complete |
| } |
| // joinBoxes is loadBoxJoin's fold over snapshots already read, so a page that |
| @@ -510,7 +521,7 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadGateway, "could not list instances") |
| return |
| } |
| - join := loadBoxJoin(ctx, s.cfg.Releases, time.Now()) |
| + join, joinComplete := loadBoxJoinComplete(ctx, s.cfg.Releases, time.Now()) |
| served := loadEdgesJoin(ctx, s.cfg.Releases) |
| params := r.URL.Query() |
| q := parseInstanceQuery(params) |
| @@ -565,19 +576,25 @@ func (s *Server) handleAPIInstanceList(w http.ResponseWriter, r *http.Request) { |
| } else { |
| end = total |
| } |
| + answer := map[string]any{ |
| + "instances": matched[start:end], "truncated": truncated, |
| + "total": total, "counts": counts, "next_cursor": next, |
| + } |
| + // Additive and only when true, so a complete join's answer is the bytes |
| + // it always was: the box join behind the rollout views could not read |
| + // every box's report, so a record with no rollout may yet have one. |
| + if !joinComplete { |
| + answer["rollout_incomplete"] = true |
| + } |
| if s.cfg.HostClaim { |
| pending, err := s.claimPendingViews(ctx, acct.Sqid, app, "instance") |
| if err != nil { |
| writeClaimResponseError(w, err) |
| return |
| } |
| - writeJSON(w, http.StatusOK, map[string]any{"instances": matched[start:end], "truncated": truncated, "total": total, "counts": counts, "next_cursor": next, "operations": pending}) |
| - return |
| + answer["operations"] = pending |
| } |
| - writeJSON(w, http.StatusOK, map[string]any{ |
| - "instances": matched[start:end], "truncated": truncated, |
| - "total": total, "counts": counts, "next_cursor": next, |
| - }) |
| + writeJSON(w, http.StatusOK, answer) |
| } |
| // handleAPIInstanceGet answers one record. A host with no record is the |
diff --git a/internal/console/rolloutjoin_test.go b/internal/console/rolloutjoin_test.go| index 4f5dea320..0dab73be3 100644 |
| --- a/internal/console/rolloutjoin_test.go |
| +++ b/internal/console/rolloutjoin_test.go |
| @@ -171,3 +171,43 @@ func TestJoinBoxesNilReportJoinsNothing(t *testing.T) { |
| t.Fatalf("nil report joined %+v", got) |
| } |
| } |
| + |
| +// Astra plan 6 R1 #4: the list says when its rollout join could not read |
| +// every box's report (rollout_incomplete), so the deploy watch never reads a |
| +// missing report as "this box does not report". The flag is additive: a |
| +// complete join leaves the key off, and the wire is as before. |
| +func TestAPIInstanceListSaysWhenTheRolloutJoinIsIncomplete(t *testing.T) { |
| + fx, token := instanceFixture(t, "code-inst-incomplete") |
| + enableInstances(t, fx, "titogo", titoManifest()) |
| + u := instancesURL(fx, "bab", "titogo") |
| + if code, _, out := rawReq(t, http.MethodPost, u, token, `{"host":"acme.go.tito.io","region":"ie"}`); code != http.StatusCreated { |
| + t.Fatalf("create = %d %s", code, out) |
| + } |
| + box := status.CollectBox(nil, status.Probes{MachineID: "box1"}, time.Now()) |
| + box.Metrics = &status.BoxMetrics{Rollout: &status.RolloutReport{Apps: map[string]status.AppRollout{"bab/titogo": {}}}} |
| + if err := status.WriteBox(context.Background(), fx.rel, box); err != nil { |
| + t.Fatal(err) |
| + } |
| + incomplete := func() (bool, bool) { |
| + t.Helper() |
| + code, _, out := rawReq(t, http.MethodGet, u, token, "") |
| + if code != http.StatusOK { |
| + t.Fatalf("list = %d %s", code, out) |
| + } |
| + var raw map[string]json.RawMessage |
| + if err := json.Unmarshal([]byte(out), &raw); err != nil { |
| + t.Fatal(err) |
| + } |
| + v, ok := raw["rollout_incomplete"] |
| + return ok, ok && string(v) == "true" |
| + } |
| + if present, _ := incomplete(); present { |
| + t.Fatal("a complete join carries rollout_incomplete") |
| + } |
| + if _, err := fx.rel.Put(context.Background(), "status/boxes/broken.json", []byte("{not json"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, set := incomplete(); !set { |
| + t.Fatal("a join that skipped an unreadable box snapshot does not say rollout_incomplete") |
| + } |
| +} |
diff --git a/internal/provision/plan9_b8_census_test.go b/internal/provision/plan9_b8_census_test.go| index 5940add21..ead7ab3b0 100644 |
| --- a/internal/provision/plan9_b8_census_test.go |
| +++ b/internal/provision/plan9_b8_census_test.go |
| @@ -584,7 +584,7 @@ var plan9B8CensusInventory = []censusEntry{ |
| {"internal/console/apiinstancemove.go", "Server.handleAPIInstanceMoveRequest", "record-writer", "key:SetRecordMove,lit:instances", "708ef4614cf5fd9c1ad382cb79da55af0b1192a235fd719765affda92aaf1668"}, |
| {"internal/console/apiinstancemove.go", "Server.registerMovesPhase1Routes", "key-site", "lit:instances", "5af589d288ba22856f631481c40cc001aac342ea74cee6ee0464ec77d0320bbc"}, |
| {"internal/console/apiinstances.go", "Server.instanceManifest", "key-site", "lit:instances", "e6160ffe84abda9163527ccb33e2ffff2109b35908adb2f3a754534325658fdf"}, |
| - {"internal/console/apiinstances.go", "Server.handleAPIInstanceList", "key-site", "lit:instances", "429011bf71564e430cdbe92fbb347d0d200c52c2cb392aeb0864a1f75bac18a1"}, |
| + {"internal/console/apiinstances.go", "Server.handleAPIInstanceList", "key-site", "lit:instances", "e6b371f7911ee013f1074b983f7d582b5b07982784b375e18304e877a5d4dd39"}, |
| {"internal/console/apiinstances.go", "Server.handleAPIInstanceCreate", "record-writer", "key:CreateRecord", "20f862446f119f25d7d72eab6edd1b7cfb1528afd5b4c352e0cb8442f5080b13"}, |
| {"internal/console/apiinstances.go", "Server.handleAPIInstanceUpdate", "record-writer", "key:UpdateRecord", "4b792c410399631eb3b38c78829df85dda045651bbc32df769767cf52f015b6d"}, |
| {"internal/console/apiscan.go", "Server.writeScanRecord", "nonrecord", "sel:Put", "c3c092f4b7be9bf6bf09d0fe1e2c9fe52892ee97e57f6c7ab964b85fe36fb5c6"}, |