amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package main
2
3import (
4 "bytes"
5 "context"
6 "errors"
7 "fmt"
8 "io"
9 "net/http"
10 "net/http/httptest"
11 "strings"
12 "sync"
13 "sync/atomic"
14 "testing"
15 "time"
16
17 "github.com/carlosframework/platform/internal/rolloutwords"
18 "github.com/carlosframework/platform/internal/status"
19)
20
21const B = "b2"
22
23func TestClassifyRollout(t *testing.T) {
24 for _, tc := range []struct {
25 name string
26 r apiRollout
27 want rolloutState
28 }{
29 {"box has not installed B", apiRollout{Installed: "b1", Serving: "b1", Awake: true}, rolloutPending},
30 {"idle scheduler, B not installed", apiRollout{Installed: "b1", Good: "b1", SchedulerIdle: true}, rolloutPending},
31 {"swapped", apiRollout{Installed: B, Serving: B, Awake: true}, rolloutSwapped},
32 {"asleep with B installed", apiRollout{Installed: B, Good: "b1"}, rolloutAsleep},
33 // A process starting or stopping is neither asleep nor swapped.
34 {"transient, idle, no serving build", apiRollout{Installed: B, Good: "b1", Transient: true, SchedulerIdle: true}, rolloutPending},
35 {"transient with B recorded as serving", apiRollout{Installed: B, Serving: B, Awake: true, Transient: true, SchedulerIdle: true}, rolloutPending},
36 {"transient but failed on B", apiRollout{Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B, Transient: true}, rolloutFailed},
37 {"queued", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
38 {"canary settling", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
39 {"failed here", apiRollout{Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth}, rolloutFailed},
40 {"hold mode, down", apiRollout{Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B}, rolloutFailed},
41 {"held by a sibling's failure", apiRollout{Installed: B, Serving: "b1", Awake: true, HeldBy: B}, rolloutHeld},
42 {"asleep and held", apiRollout{Installed: B, Good: "b1", HeldBy: B}, rolloutHeld},
43 {"an older build's failure is not B's", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptFailed, AttemptBuild: "b0"}, rolloutSwapped},
44 // R8: an open attempt is pending whatever awake says, and B serving
45 // is not swapped until the canary's settle closes.
46 {"R8 canary serving B, settle open", apiRollout{Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
47 {"R8 first wake in flight", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptSettling, AttemptBuild: B}, rolloutPending},
48 {"R8 asleep, queued", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptQueued, AttemptBuild: B}, rolloutPending},
49 {"R8 asleep, running", apiRollout{Installed: B, Good: "b1", Attempt: status.AttemptRunning, AttemptBuild: B}, rolloutPending},
50 // R6: a box that cannot arm reports no per-host state: its awake
51 // false is meaningless, so B installed there is not asleep.
52 {"R6 refusing box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}, rolloutPending},
53 {"R6 declined box, B installed", apiRollout{Box: "x1", Installed: B, Unavailable: "generations-off"}, rolloutPending},
54 } {
55 r := tc.r
56 if got := classifyRollout(&r, B); got != tc.want {
57 t.Errorf("%s: %s, want %s", tc.name, got, tc.want)
58 }
59 }
60}
61
62// sweep is one scripted answer of the instance list.
63type sweep struct {
64 recs []apiInstanceRecord
65 truncated bool
66 incomplete bool
67 err error
68}
69
70// scripted answers each list call with the next sweep, repeating the last.
71func scripted(calls *atomic.Int64, steps ...sweep) func(time.Time) (watchList, error) {
72 return func(time.Time) (watchList, error) {
73 i := int(calls.Add(1)) - 1
74 s := steps[min(i, len(steps)-1)]
75 return watchList{recs: s.recs, truncated: s.truncated, incomplete: s.incomplete}, s.err
76 }
77}
78
79// sweeps answers each list call with the next set, repeating the last; no
80// list is truncated.
81func sweeps(calls *atomic.Int64, sets ...[]apiInstanceRecord) func(time.Time) (watchList, error) {
82 steps := make([]sweep, 0, len(sets))
83 for _, s := range sets {
84 steps = append(steps, sweep{recs: s})
85 }
86 return scripted(calls, steps...)
87}
88
89func rec(host string, r *apiRollout) apiInstanceRecord {
90 return apiInstanceRecord{Host: host, Rollout: r}
91}
92
93func targetsFor(hosts ...string) []watchTarget {
94 var out []watchTarget
95 for _, h := range hosts {
96 out = append(out, watchTarget{Host: h, Origin: "http://127.0.0.1:1"})
97 }
98 return out
99}
100
101// noURL fails the test if a host is polled over HTTP: a reported host must
102// never be requested, because a request wakes an asleep instance.
103func noURL(t *testing.T) func(watchTarget, int64) (bool, string) {
104 var told sync.Map
105 return func(tg watchTarget, _ int64) (bool, string) {
106 if _, dup := told.LoadOrStore(tg.Host, true); !dup {
107 t.Errorf("%s was polled over HTTP", tg.Host)
108 }
109 return false, ""
110 }
111}
112
113// pollLog records which hosts were polled over HTTP and answers live as told.
114type pollLog struct {
115 mu sync.Mutex
116 polled []string
117}
118
119func (p *pollLog) check(live map[string]bool) func(watchTarget, int64) (bool, string) {
120 return func(tg watchTarget, _ int64) (bool, string) {
121 p.mu.Lock()
122 defer p.mu.Unlock()
123 p.polled = append(p.polled, tg.Host)
124 if live[tg.Host] {
125 return true, ""
126 }
127 return false, "still serving b1"
128 }
129}
130
131func (p *pollLog) hosts() map[string]int {
132 p.mu.Lock()
133 defer p.mu.Unlock()
134 out := map[string]int{}
135 for _, h := range p.polled {
136 out[h]++
137 }
138 return out
139}
140
141func testWatch(w io.Writer, list func(time.Time) (watchList, error), check func(watchTarget, int64) (bool, string)) rolloutWatch {
142 return rolloutWatch{w: w, app: "shop", ver: B, channel: "edge", timeout: 300 * time.Millisecond,
143 interval: 5 * time.Millisecond, list: list, check: check}
144}
145
146func TestWatchRolloutFollowsTheBoxToDone(t *testing.T) {
147 var out bytes.Buffer
148 var calls atomic.Int64
149 list := sweeps(&calls,
150 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
151 rec("b.test", &apiRollout{Box: "x1", Installed: "b1", Good: "b1", SchedulerIdle: true})},
152 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
153 rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
154 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
155 rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
156 )
157 if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
158 t.Fatalf("watch: %v\n%s", err, out.String())
159 }
160 for _, want := range []string{
161 fmt.Sprintf(watchIntro, B),
162 watchWordSwapped + " https://a.test",
163 watchWordInstalled + " https://b.test (" + watchAsleepTail + ")",
164 fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountSwapped, 1)+", "+fmt.Sprintf(watchCountAsleep, 1)),
165 } {
166 if !strings.Contains(out.String(), want) {
167 t.Errorf("output lacks %q:\n%s", want, out.String())
168 }
169 }
170 if calls.Load() < 3 {
171 t.Errorf("done after %d sweeps: an open settle must hold the box", calls.Load())
172 }
173}
174
175// A host whose process is starting or stopping stays pending through a sweep,
176// idle scheduler or not, and completes once it reports asleep or swapped.
177func TestWatchRolloutATransientHostStaysPendingUntilItSettles(t *testing.T) {
178 for _, tc := range []struct {
179 name string
180 final apiRollout
181 want string
182 }{
183 {"asleep", apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}, watchWordInstalled + " https://a.test (" + watchAsleepTail + ")"},
184 {"swapped", apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}, watchWordSwapped + " https://a.test"},
185 } {
186 t.Run(tc.name, func(t *testing.T) {
187 var out bytes.Buffer
188 var calls atomic.Int64
189 list := sweeps(&calls,
190 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", Transient: true, SchedulerIdle: true})},
191 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", Transient: true, SchedulerIdle: true})},
192 []apiInstanceRecord{rec("a.test", &tc.final)},
193 )
194 if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test")); err != nil {
195 t.Fatalf("watch: %v\n%s", err, out.String())
196 }
197 if calls.Load() < 3 {
198 t.Fatalf("done after %d sweeps, while the host was still starting or stopping", calls.Load())
199 }
200 if !strings.Contains(out.String(), tc.want) {
201 t.Errorf("output lacks %q:\n%s", tc.want, out.String())
202 }
203 })
204 }
205}
206
207// R8: a canary already serving B is not swapped while its settle is open.
208func TestWatchRolloutCanaryServingBWaitsForItsSettle(t *testing.T) {
209 var out bytes.Buffer
210 var calls atomic.Int64
211 list := sweeps(&calls,
212 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B, SchedulerIdle: false})},
213 []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})},
214 )
215 err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test"))
216 if err == nil || !strings.HasPrefix(err.Error(), fmt.Sprintf(watchFailedSummary, B, 1, 0, "shop", B, "edge")) {
217 t.Fatalf("err = %v\n%s", err, out.String())
218 }
219 if strings.Contains(out.String(), watchWordSwapped) {
220 t.Fatalf("a canary in its settle was counted swapped:\n%s", out.String())
221 }
222}
223
224// A host that is final keeps being refreshed: here the sleeper is final
225// (installed, asleep) from the first sweep, while its box's scheduler is
226// still busy, and the box goes idle a sweep later. A loop that skips final
227// hosts keeps the first report's SchedulerIdle=false for ever, so boxDone
228// never holds and a finished deploy times out. The host prints once.
229func TestWatchRolloutRefreshesAFinalHostUntilTheBoxIsDone(t *testing.T) {
230 var out bytes.Buffer
231 var calls atomic.Int64
232 list := sweeps(&calls,
233 []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
234 []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"})},
235 []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})},
236 )
237 if err := testWatch(&out, list, noURL(t)).run(targetsFor("b.test")); err != nil {
238 t.Fatalf("watch: %v\n%s", err, out.String())
239 }
240 line := watchWordInstalled + " https://b.test (" + watchAsleepTail + ")"
241 if got := strings.Count(out.String(), line); got != 1 {
242 t.Fatalf("%q printed %d times, want once:\n%s", line, got, out.String())
243 }
244 if calls.Load() < 3 {
245 t.Fatalf("done after %d sweeps, before the box's scheduler went idle", calls.Load())
246 }
247}
248
249// Review Focus 2.
250func TestWatchRolloutNoOpDeployEndsAtOnce(t *testing.T) {
251 var calls atomic.Int64
252 list := sweeps(&calls, []apiInstanceRecord{
253 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
254 rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: B, SchedulerIdle: true}),
255 })
256 if err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
257 t.Fatal(err)
258 }
259 if calls.Load() != 1 {
260 t.Fatalf("an already-converged deploy took %d sweeps, want 1", calls.Load())
261 }
262}
263
264func TestWatchRolloutIdleSchedulerBeforeInstallIsPending(t *testing.T) {
265 var calls atomic.Int64
266 list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true})})
267 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
268 if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchTimeout, B)) || !strings.Contains(err.Error(), fmt.Sprintf(watchPendingOther, "b1")) {
269 t.Fatalf("err = %v", err)
270 }
271}
272
273// Review Focus 4: a newer build replaced the watched one.
274func TestWatchRolloutNewerBuildNamesWhatIsInstalled(t *testing.T) {
275 var calls atomic.Int64
276 list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true})})
277 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
278 if err == nil || !strings.Contains(err.Error(), "a.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
279 t.Fatalf("err = %v", err)
280 }
281}
282
283func TestWatchRolloutOpenSettleHoldsTheWatch(t *testing.T) {
284 var calls atomic.Int64
285 list := sweeps(&calls, []apiInstanceRecord{
286 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true}),
287 rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1"}),
288 })
289 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
290 if err == nil || !strings.Contains(err.Error(), fmt.Sprintf(watchBoxBusy, "x1")) {
291 t.Fatalf("err = %v", err)
292 }
293}
294
295// R7: the host the build failed on is failed; a sibling its record holds
296// back is held.
297func TestWatchRolloutReportsFailureAndNamesPromoteRetry(t *testing.T) {
298 var out bytes.Buffer
299 var calls atomic.Int64
300 list := sweeps(&calls, []apiInstanceRecord{
301 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptFailed, AttemptBuild: B, HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
302 rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
303 rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", HeldBy: B, Reason: status.ReasonHealth, SchedulerIdle: true}),
304 })
305 err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "c.test"))
306 if err == nil || err.Error() != fmt.Sprintf(watchFailedSummary, B, 1, 2, "shop", B, "edge") {
307 t.Fatalf("err = %v", err)
308 }
309 for _, want := range []string{
310 watchWordFailed + " https://a.test: " + rolloutwords.Reason(status.ReasonHealth),
311 watchWordHeld + " https://b.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
312 watchWordHeld + " https://c.test: " + fmt.Sprintf(watchHeldTail, "b1", B),
313 fmt.Sprintf(watchBox, "x1", fmt.Sprintf(watchCountFailed, 1)+", "+fmt.Sprintf(watchCountHeld, 2)),
314 } {
315 if !strings.Contains(out.String(), want) {
316 t.Errorf("output lacks %q:\n%s", want, out.String())
317 }
318 }
319 if strings.Contains(err.Error(), watchHoldDownSummary[:20]) {
320 t.Errorf("an overlap failure named the hold-down rollback: %v", err)
321 }
322}
323
324// R7: a hold failure (attempt down for B) is failed, and the summary adds
325// the hold-down sentence.
326func TestWatchRolloutHoldDownNamesRollback(t *testing.T) {
327 var calls atomic.Int64
328 list := sweeps(&calls, []apiInstanceRecord{
329 rec("a.test", &apiRollout{Box: "x1", Installed: B, Attempt: status.AttemptDown, AttemptBuild: B, HeldBy: B, Reason: status.ReasonExit, SchedulerIdle: true}),
330 rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, HeldBy: B, SchedulerIdle: true}),
331 })
332 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
333 want := fmt.Sprintf(watchFailedSummary, B, 1, 1, "shop", B, "edge") + "\n" + fmt.Sprintf(watchHoldDownSummary, "shop")
334 if err == nil || err.Error() != want {
335 t.Fatalf("err = %v\nwant %s", err, want)
336 }
337}
338
339// Review Focus 3 and R1: a box that stops reporting never falls back to URL
340// polling.
341func TestWatchRolloutDarkBoxNeverPollsOverHTTP(t *testing.T) {
342 var calls atomic.Int64
343 list := sweeps(&calls,
344 []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true})},
345 []apiInstanceRecord{rec("a.test", nil)},
346 )
347 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
348 if err == nil || !strings.Contains(err.Error(), "a.test: "+watchPendingDark) {
349 t.Fatalf("err = %v", err)
350 }
351}
352
353// A final state survives a missing observation, never an explicit later
354// report (Astra plan 6 R1 #3; Review Focus 4). a.test swaps and its box goes
355// dark: it stays swapped. d.test swaps and its box then reports b3 installed
356// and serving: B no longer runs there, so d.test is pending again, named by
357// what is installed, and the watch must not succeed for B once x2's canary
358// settles.
359func TestWatchRolloutFinalSurvivesADarkBoxButNotANewerBuild(t *testing.T) {
360 var out bytes.Buffer
361 var calls atomic.Int64
362 list := sweeps(&calls,
363 []apiInstanceRecord{
364 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
365 rec("d.test", &apiRollout{Box: "x3", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
366 rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
367 },
368 []apiInstanceRecord{
369 rec("a.test", nil),
370 rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
371 rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}),
372 },
373 []apiInstanceRecord{
374 rec("a.test", nil),
375 rec("d.test", &apiRollout{Box: "x3", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
376 rec("b.test", &apiRollout{Box: "x2", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
377 },
378 )
379 err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test", "d.test"))
380 if err == nil {
381 t.Fatalf("the watch succeeded for %s though d.test's box reports b3:\n%s", B, out.String())
382 }
383 if !strings.Contains(err.Error(), "\n d.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
384 t.Errorf("timeout does not name d.test by what is installed: %v", err)
385 }
386 for _, h := range []string{"a.test", "b.test"} {
387 if strings.Contains(err.Error(), "\n "+h+": ") {
388 t.Errorf("timeout names %s, which is final: %v", h, err)
389 }
390 if got := strings.Count(out.String(), watchWordSwapped+" https://"+h+"\n"); got != 1 {
391 t.Errorf("%s printed swapped %d times:\n%s", h, got, out.String())
392 }
393 }
394}
395
396// A host watched by URL that a later report follows is reclassified by the
397// report (Astra plan 6 R1 #3). c.test answers live by URL while b.test's
398// canary settles; the next list reports c.test's box with b3 installed, so
399// the watch must not count c.test done for B.
400func TestWatchRolloutLaterReportReclassifiesAURLLiveHost(t *testing.T) {
401 var out bytes.Buffer
402 var calls atomic.Int64
403 var pl pollLog
404 settling := rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B})
405 list := sweeps(&calls,
406 []apiInstanceRecord{settling, rec("c.test", nil)},
407 []apiInstanceRecord{settling, rec("c.test", &apiRollout{Box: "x2", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true})},
408 []apiInstanceRecord{
409 rec("b.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
410 rec("c.test", &apiRollout{Box: "x2", Installed: "b3", Serving: "b3", Awake: true, SchedulerIdle: true}),
411 },
412 )
413 err := testWatch(&out, list, pl.check(map[string]bool{"c.test": true})).run(targetsFor("b.test", "c.test"))
414 if err == nil || !strings.Contains(err.Error(), "\n c.test: "+fmt.Sprintf(watchPendingOther, "b3")) {
415 t.Fatalf("err = %v\n%s", err, out.String())
416 }
417 if got := pl.hosts()["c.test"]; got != 1 {
418 t.Errorf("c.test polled %d times, want once (before its report)", got)
419 }
420}
421
422// R3: absence from a truncated list keeps a host where it was. b.test is
423// missing from the first (truncated) sweep and the list then fails once;
424// it is never polled over HTTP, and its report, when it comes, decides it.
425// a.test is missing from a later truncated sweep and keeps its state.
426func TestWatchRolloutTruncatedListNeverSwitchesAHostToURL(t *testing.T) {
427 var out bytes.Buffer
428 var calls atomic.Int64
429 busyA := rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B})
430 list := scripted(&calls,
431 sweep{recs: []apiInstanceRecord{busyA}, truncated: true},
432 sweep{err: errors.New("instances list: 429")},
433 sweep{recs: []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true})}, truncated: true},
434 sweep{recs: []apiInstanceRecord{
435 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
436 rec("b.test", &apiRollout{Box: "x2", Installed: B, Good: "b1", SchedulerIdle: true}),
437 }},
438 )
439 if err := testWatch(&out, list, noURL(t)).run(targetsFor("a.test", "b.test")); err != nil {
440 t.Fatalf("watch: %v\n%s", err, out.String())
441 }
442 if calls.Load() < 4 {
443 t.Fatalf("done after %d sweeps: a.test was still settling when the list cut it off", calls.Load())
444 }
445}
446
447// R3: a host no truncated list ever names stays unknown, and the timeout
448// says why.
449func TestWatchRolloutAlwaysTruncatedNamesTheUnlistedHost(t *testing.T) {
450 var calls atomic.Int64
451 list := scripted(&calls, sweep{recs: []apiInstanceRecord{
452 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
453 }, truncated: true})
454 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test", "b.test"))
455 if err == nil || !strings.Contains(err.Error(), "b.test: "+watchPendingUnlisted) {
456 t.Fatalf("err = %v", err)
457 }
458}
459
460// R4: a host its box did not report on at first (created since the box's
461// last snapshot) is watched by URL until a report comes, then by the report.
462func TestWatchRolloutALateReportMovesAHostOffURL(t *testing.T) {
463 var out bytes.Buffer
464 var calls atomic.Int64
465 var pl pollLog
466 list := sweeps(&calls,
467 []apiInstanceRecord{
468 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
469 rec("c.test", nil),
470 },
471 []apiInstanceRecord{
472 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
473 rec("c.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}),
474 },
475 )
476 if err := testWatch(&out, list, pl.check(nil)).run(targetsFor("a.test", "c.test")); err != nil {
477 t.Fatalf("watch: %v\n%s", err, out.String())
478 }
479 if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 {
480 t.Errorf("polled %v, want c.test once (before its report), a.test never", got)
481 }
482 if !strings.Contains(out.String(), watchWordInstalled+" https://c.test ("+watchAsleepTail+")") {
483 t.Errorf("c.test not followed through its report:\n%s", out.String())
484 }
485}
486
487func TestWatchRolloutMixedFleetPollsOnlyTheUnreported(t *testing.T) {
488 var out bytes.Buffer
489 var calls atomic.Int64
490 var pl pollLog
491 list := sweeps(&calls, []apiInstanceRecord{
492 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
493 rec("c.test", nil),
494 })
495 rw := testWatch(&out, list, pl.check(map[string]bool{"c.test": true}))
496 onURL := 0
497 rw.onURL = func() { onURL++ }
498 if err := rw.run(targetsFor("a.test", "c.test")); err != nil {
499 t.Fatal(err)
500 }
501 if got := pl.hosts(); len(got) != 1 || got["c.test"] != 1 || onURL != 1 || !strings.Contains(out.String(), "live https://c.test") {
502 t.Fatalf("polled %v, onURL %d, output:\n%s", got, onURL, out.String())
503 }
504}
505
506// R6: a box that refuses every managed start holds its instances pending,
507// never polled, and the timeout names why in the box's words.
508func TestWatchRolloutRefusingBoxIsPendingWithItsReason(t *testing.T) {
509 var calls atomic.Int64
510 list := sweeps(&calls, []apiInstanceRecord{
511 rec("a.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-release-key"}),
512 })
513 err := testWatch(io.Discard, list, noURL(t)).run(targetsFor("a.test"))
514 if err == nil || !strings.Contains(err.Error(), "a.test: "+rolloutwords.Unavailable("no-release-key")) {
515 t.Fatalf("err = %v", err)
516 }
517}
518
519// R6: a box that declined overlap runs today's stop-then-start paths, so its
520// instances are watched by URL, like a flag-off box's.
521func TestWatchRolloutDeclinedBoxIsWatchedByURL(t *testing.T) {
522 var out bytes.Buffer
523 var calls atomic.Int64
524 var pl pollLog
525 list := sweeps(&calls, []apiInstanceRecord{
526 rec("a.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true, SchedulerIdle: true}),
527 rec("c.test", &apiRollout{Box: "x2", Installed: B, Unavailable: "generations-off"}),
528 })
529 if err := testWatch(&out, list, pl.check(map[string]bool{"c.test": true})).run(targetsFor("a.test", "c.test")); err != nil {
530 t.Fatalf("watch: %v\n%s", err, out.String())
531 }
532 if got := pl.hosts(); got["c.test"] != 1 || got["a.test"] != 0 || !strings.Contains(out.String(), "live https://c.test") {
533 t.Fatalf("polled %v, output:\n%s", got, out.String())
534 }
535}
536
537// R5: the instance list is read on its own cadence (5 s in production), not
538// on every URL poll.
539func TestWatchRolloutListsOnItsOwnCadence(t *testing.T) {
540 var calls atomic.Int64
541 var pl pollLog
542 list := sweeps(&calls, []apiInstanceRecord{
543 rec("a.test", &apiRollout{Box: "x1", Installed: "b1", Serving: "b1", Awake: true, SchedulerIdle: true}),
544 rec("c.test", nil),
545 })
546 rw := testWatch(io.Discard, list, pl.check(nil))
547 rw.timeout, rw.listEvery = 80*time.Millisecond, time.Hour
548 if err := rw.run(targetsFor("a.test", "c.test")); err == nil {
549 t.Fatal("watch ended, want a timeout")
550 }
551 if calls.Load() != 1 || pl.hosts()["c.test"] < 3 {
552 t.Fatalf("%d lists for %d URL polls: the list must keep its own cadence", calls.Load(), pl.hosts()["c.test"])
553 }
554}
555
556func TestDeployWatchCadence(t *testing.T) {
557 rw := newDeployWatch(nil, nil, apiAcct{Seg: "acme"}, "shop", B, "edge", "", time.Minute)
558 if rw.listEvery != 5*time.Second || rw.interval != 2*time.Second {
559 t.Fatalf("list every %s, poll every %s: want 5s and 2s", rw.listEvery, rw.interval)
560 }
561}
562
563// R10: with no box reporting, the watch is today's watchVersion, byte for
564// byte, its timeout report included.
565func TestWatchRolloutWithNoReportsIsTheOldWatch(t *testing.T) {
566 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
567 w.Header().Set("X-Carlos-Version", B)
568 }))
569 defer srv.Close()
570 var out bytes.Buffer
571 var calls atomic.Int64
572 onURL := 0
573 rw := testWatch(&out, sweeps(&calls, []apiInstanceRecord{rec("x.test", nil)}), nil)
574 rw.onURL = func() { onURL++ }
575 if err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}}); err != nil {
576 t.Fatal(err)
577 }
578 if onURL != 1 || !strings.Contains(out.String(), "live https://x.test") || strings.Contains(out.String(), watchWordSwapped) {
579 t.Fatalf("onURL %d, output:\n%s", onURL, out.String())
580 }
581}
582
583func TestWatchRolloutWithNoReportsTimesOutAsTheOldWatch(t *testing.T) {
584 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
585 w.Header().Set("X-Carlos-Version", "b1")
586 }))
587 defer srv.Close()
588 targets := []watchTarget{{Host: "x.test", Origin: srv.URL}, {Host: "y.test", Origin: srv.URL}}
589 diag := &watchDiagnostics{logsHint: "carlos logs --app shop"}
590 var oldOut bytes.Buffer
591 oldErr := watchVersion(&oldOut, targets, B, 30*time.Millisecond, 5*time.Millisecond, diag)
592 for name, recs := range map[string][]apiInstanceRecord{
593 "flag off": {rec("x.test", nil), rec("y.test", nil)},
594 // R6: every box declined overlap.
595 "declined": {rec("x.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "no-cgroup-delegation"}), rec("y.test", nil)},
596 // A claims-on box declines overlap too (#272): URL-watched like any declined box.
597 "claims-on": {rec("x.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "host-claims-on"}), rec("y.test", nil)},
598 // A report on an instance the deploy does not target is not one.
599 "untargeted report": {rec("x.test", nil), rec("y.test", nil), rec("z.test", &apiRollout{Box: "x1", Installed: B, Serving: B, Awake: true})},
600 } {
601 var out bytes.Buffer
602 var calls atomic.Int64
603 rw := testWatch(&out, sweeps(&calls, recs), nil)
604 rw.timeout, rw.diag = 30*time.Millisecond, diag
605 err := rw.run(targets)
606 if err == nil || oldErr == nil || err.Error() != oldErr.Error() || out.String() != oldOut.String() {
607 t.Errorf("%s: err = %v\nwant %v\nout %q want %q", name, err, oldErr, out.String(), oldOut.String())
608 }
609 }
610}
611
612// R9: the until-about form only for a held_until in the future; otherwise
613// today's line, byte for byte.
614func TestDeployHeldLine(t *testing.T) {
615 now := time.Date(2026, 10, 10, 12, 0, 0, 0, time.UTC)
616 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"
617 future := now.Add(3 * time.Hour)
618 for _, tc := range []struct {
619 name, until, want string
620 }{
621 {"absent", "", today},
622 {"past", now.Add(-time.Minute).Format(time.RFC3339), today},
623 {"now", now.Format(time.RFC3339), today},
624 {"unparseable", "soon", today},
625 {"future", future.Format(time.RFC3339), fmt.Sprintf(deployHeldUntil, "shop", "stable", future.Local().Format("Mon 2 Jan 15:04 MST"))},
626 } {
627 if got := deployHeldLine("shop", "stable", tc.until, now); got != tc.want {
628 t.Errorf("%s: %q\nwant %q", tc.name, got, tc.want)
629 }
630 }
631}
632
633// hitServer is an instance's origin that counts the HTTP requests reaching
634// it, so a test sees a poll from either path: the rollout watch's check or a
635// fallback to watchVersion, which dials the origin itself.
636func hitServer(t *testing.T) (string, *atomic.Int64) {
637 var hits atomic.Int64
638 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
639 hits.Add(1)
640 w.Header().Set("X-Carlos-Version", B)
641 }))
642 t.Cleanup(srv.Close)
643 return srv.URL, &hits
644}
645
646// Fix round 1: a failed first list is retried, not a reason to URL-poll a
647// fleet whose boxes report (that wakes the sleeper here).
648func TestWatchRolloutFirstListErrorIsRetried(t *testing.T) {
649 origin, hits := hitServer(t)
650 var calls atomic.Int64
651 list := scripted(&calls,
652 sweep{err: errors.New("instances list: 429")},
653 sweep{recs: []apiInstanceRecord{rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})}},
654 )
655 rw := testWatch(io.Discard, list, noURL(t))
656 onURL := 0
657 rw.onURL = func() { onURL++ }
658 if err := rw.run([]watchTarget{{Host: "b.test", Origin: origin}}); err != nil {
659 t.Fatal(err)
660 }
661 if hits.Load() != 0 || onURL != 0 || calls.Load() != 2 {
662 t.Fatalf("%d HTTP polls, onURL %d, %d lists: want 0, 0, 2", hits.Load(), onURL, calls.Load())
663 }
664}
665
666// Astra plan 6 R1 #4 (reverses fix round 1's three-tries fallback): failed
667// lists are no evidence that the boxes do not report, so they are retried
668// until a list answers. Here the fourth does, with a report: the sleeper is
669// never requested.
670func TestWatchRolloutFailedFirstListsNeverFallBack(t *testing.T) {
671 origin, hits := hitServer(t)
672 var calls atomic.Int64
673 var out bytes.Buffer
674 failed := sweep{err: &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("instances list: 503")}}
675 list := scripted(&calls, failed, failed, failed,
676 sweep{recs: []apiInstanceRecord{rec("x.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true})}},
677 )
678 rw := testWatch(&out, list, nil)
679 onURL := 0
680 rw.onURL = func() { onURL++ }
681 if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil {
682 t.Fatalf("watch: %v\n%s", err, out.String())
683 }
684 if calls.Load() != 4 || onURL != 0 || hits.Load() != 0 || !strings.Contains(out.String(), watchWordInstalled+" https://x.test") {
685 t.Fatalf("%d lists, onURL %d, %d HTTP polls, output:\n%s", calls.Load(), onURL, hits.Load(), out.String())
686 }
687}
688
689// Astra plan 6 R1 #4: a list that keeps failing with a 5xx, a 429 or a
690// network error is retried until the deadline. No origin is requested, and
691// the timeout names the list failure.
692func TestWatchRolloutListFailuresTimeOutNamingTheList(t *testing.T) {
693 for name, lerr := range map[string]error{
694 "503": &listStatusError{status: http.StatusServiceUnavailable, err: errors.New("could not list instances")},
695 "429": &listStatusError{status: http.StatusTooManyRequests, err: errors.New("too many requests")},
696 // Astra plan 6 R2 #4: the deploy just authenticated with this
697 // token, so a 401 says nothing about the boxes.
698 "401": &listStatusError{status: http.StatusUnauthorized, err: errors.New("not signed in")},
699 "network": errors.New("dial tcp: connection refused"),
700 } {
701 origin, hits := hitServer(t)
702 var calls atomic.Int64
703 rw := testWatch(io.Discard, scripted(&calls, sweep{err: lerr}), nil)
704 onURL := 0
705 rw.onURL = func() { onURL++ }
706 err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
707 if hits.Load() != 0 || onURL != 0 {
708 t.Errorf("%s: %d HTTP polls, onURL %d: want none", name, hits.Load(), onURL)
709 }
710 want := fmt.Sprintf(watchTimeout, B) + "\n " + watchListFailed
711 if err == nil || !strings.HasPrefix(err.Error(), want) || !strings.Contains(err.Error(), lerr.Error()) {
712 t.Errorf("%s: err = %v\nwant it to start %q and carry the list's error", name, err, want)
713 }
714 if calls.Load() < 2 {
715 t.Errorf("%s: %d lists: a failed list must be retried", name, calls.Load())
716 }
717 }
718}
719
720// Astra plan 6 R1 #4, R2 #4: a 404 is a console with no list for the app,
721// and a 403 a principal that may deploy but not read the list (a
722// Publish-only token): either way the watch is today's, as before.
723func TestWatchRolloutList404And403FallBack(t *testing.T) {
724 for _, lerr := range []*listStatusError{
725 {status: http.StatusNotFound, err: notFoundErr()},
726 {status: http.StatusForbidden, err: errors.New("forbidden")},
727 } {
728 origin, hits := hitServer(t)
729 var calls atomic.Int64
730 var out bytes.Buffer
731 rw := testWatch(&out, scripted(&calls, sweep{err: lerr}), nil)
732 onURL := 0
733 rw.onURL = func() { onURL++ }
734 if err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}}); err != nil {
735 t.Fatalf("%d: %v", lerr.status, err)
736 }
737 if calls.Load() != 1 || onURL != 1 || hits.Load() == 0 || out.String() != "live https://x.test\n" {
738 t.Fatalf("%d: %d lists, onURL %d, %d HTTP polls, output %q", lerr.status, calls.Load(), onURL, hits.Load(), out.String())
739 }
740 }
741}
742
743// Astra plan 6 R1 #4: a list whose rollout join was incomplete is no
744// evidence that a host has no report: it is not URL-polled, and the timeout
745// says why its state is unknown.
746func TestWatchRolloutIncompleteJoinIsNotNoReports(t *testing.T) {
747 origin, hits := hitServer(t)
748 var calls atomic.Int64
749 rw := testWatch(io.Discard, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil)}, incomplete: true}), nil)
750 onURL := 0
751 rw.onURL = func() { onURL++ }
752 err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
753 if hits.Load() != 0 || onURL != 0 {
754 t.Fatalf("%d HTTP polls, onURL %d: want none", hits.Load(), onURL)
755 }
756 if err == nil || !strings.Contains(err.Error(), "\n x.test: "+watchPendingUnread) {
757 t.Fatalf("err = %v", err)
758 }
759}
760
761// Astra plan 6 R1 #7: an app-wide truncated list on which every target is
762// listed with no report is the old watch, byte for byte, on success and on
763// timeout.
764func TestWatchRolloutTruncatedListWithEveryTargetUnreportedIsTheOldWatch(t *testing.T) {
765 for _, served := range []string{B, "b1"} {
766 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
767 w.Header().Set("X-Carlos-Version", served)
768 }))
769 targets := []watchTarget{{Host: "x.test", Origin: srv.URL}, {Host: "y.test", Origin: srv.URL}}
770 diag := &watchDiagnostics{logsHint: "carlos logs --app shop"}
771 var oldOut bytes.Buffer
772 oldErr := watchVersion(&oldOut, targets, B, 30*time.Millisecond, 5*time.Millisecond, diag)
773 var out bytes.Buffer
774 var calls atomic.Int64
775 // The production check, so a watch that wrongly stays on the reports
776 // path prints what it would print, rather than panicking.
777 client := &http.Client{Timeout: time.Second}
778 rw := testWatch(&out, scripted(&calls, sweep{recs: []apiInstanceRecord{rec("x.test", nil), rec("y.test", nil)}, truncated: true}),
779 func(t watchTarget, n int64) (bool, string) { return checkTarget(client, t, B, n) })
780 rw.timeout, rw.diag = 30*time.Millisecond, diag
781 err := rw.run(targets)
782 srv.Close()
783 if fmt.Sprint(err) != fmt.Sprint(oldErr) || out.String() != oldOut.String() {
784 t.Errorf("serving %s: err = %v\nwant %v\nout %q\nwant %q", served, err, oldErr, out.String(), oldOut.String())
785 }
786 }
787}
788
789// Astra plan 6 R1 #2: one deadline governs discovery. A list that keeps
790// failing at the production cadence (5 s between lists) does not hold a
791// 50 ms watch for ten seconds.
792func TestWatchRolloutTinyTimeoutHonouredThroughDiscovery(t *testing.T) {
793 var calls atomic.Int64
794 rw := testWatch(io.Discard, scripted(&calls, sweep{err: &listStatusError{status: http.StatusBadGateway, err: errors.New("could not list instances")}}), nil)
795 rw.timeout, rw.listEvery, rw.interval = 50*time.Millisecond, deployWatchListEvery, deployWatchPollEvery
796 start := time.Now()
797 err := rw.run(targetsFor("x.test"))
798 if took := time.Since(start); took > time.Second {
799 t.Fatalf("a 50ms watch took %s through a failing list", took)
800 }
801 if err == nil || !strings.Contains(err.Error(), watchListFailed) {
802 t.Fatalf("err = %v", err)
803 }
804}
805
806// Astra plan 6 R1 #2: the fallback gets what is left of the deadline, not
807// the whole timeout again. The list takes 300 ms to answer 404 inside a
808// 400 ms watch; today's watch then has about 100 ms, not 400.
809func TestWatchRolloutFallbackGetsTheRemainingTime(t *testing.T) {
810 srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
811 w.Header().Set("X-Carlos-Version", "b1")
812 }))
813 defer srv.Close()
814 rw := testWatch(io.Discard, func(time.Time) (watchList, error) {
815 time.Sleep(300 * time.Millisecond)
816 return watchList{}, &listStatusError{status: http.StatusNotFound, err: notFoundErr()}
817 }, nil)
818 rw.timeout = 400 * time.Millisecond
819 start := time.Now()
820 err := rw.run([]watchTarget{{Host: "x.test", Origin: srv.URL}})
821 if took := time.Since(start); took > 600*time.Millisecond {
822 t.Fatalf("a 400ms watch took %s: the fallback was given the whole timeout again", took)
823 }
824 if err == nil || !strings.Contains(err.Error(), "timed out waiting for x.test to serve "+B) {
825 t.Fatalf("err = %v", err)
826 }
827}
828
829// The production list reader, against a console: a 503 is retried to the
830// deadline with no origin request; a 404 with no JSON error falls back; and
831// rollout_incomplete is decoded, so an incomplete join is never read as no
832// reports.
833func TestDeployWatchReadsTheConsolesList(t *testing.T) {
834 for _, tc := range []struct {
835 name string
836 answer func(w http.ResponseWriter)
837 wantHits bool
838 wantErr string
839 }{
840 {"503", func(w http.ResponseWriter) {
841 w.WriteHeader(http.StatusServiceUnavailable)
842 io.WriteString(w, `{"error":"could not list instances"}`)
843 }, false, watchListFailed + ": could not list instances"},
844 {"404", func(w http.ResponseWriter) { w.WriteHeader(http.StatusNotFound) }, true, ""},
845 {"401", func(w http.ResponseWriter) {
846 w.WriteHeader(http.StatusUnauthorized)
847 io.WriteString(w, `{"error":"not signed in"}`)
848 }, false, watchListFailed + ": not signed in"},
849 {"incomplete join", func(w http.ResponseWriter) {
850 io.WriteString(w, `{"instances":[{"host":"x.test"}],"truncated":false,"rollout_incomplete":true}`)
851 }, false, "x.test: " + watchPendingUnread},
852 } {
853 console := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
854 if !strings.HasSuffix(r.URL.Path, "/instances") {
855 http.NotFound(w, r)
856 return
857 }
858 tc.answer(w)
859 }))
860 origin, hits := hitServer(t)
861 sess := &apiSession{c: newConsoleClient(console.URL, "tok"), console: console.URL}
862 rw := newDeployWatch(context.Background(), sess, apiAcct{Seg: "acme"}, "shop", B, "edge", "", 300*time.Millisecond)
863 rw.w, rw.onURL, rw.diag = io.Discard, nil, nil
864 start := time.Now()
865 err := rw.run([]watchTarget{{Host: "x.test", Origin: origin}})
866 took := time.Since(start)
867 console.Close()
868 if (hits.Load() > 0) != tc.wantHits {
869 t.Errorf("%s: %d origin requests", tc.name, hits.Load())
870 }
871 if tc.wantErr == "" && err != nil || tc.wantErr != "" && (err == nil || !strings.Contains(err.Error(), tc.wantErr)) {
872 t.Errorf("%s: err = %v, want %q", tc.name, err, tc.wantErr)
873 }
874 if took > 2*time.Second {
875 t.Errorf("%s: a 300ms watch took %s", tc.name, took)
876 }
877 }
878}
879
880// Fix round 1 (R3 on the first sweep): a truncated first list with no
881// report among the listed targets is no evidence that no box reports.
882// c.test (listed, unreported) is watched by URL; b.test (cut off, then
883// reported asleep) is never requested.
884func TestWatchRolloutTruncatedFirstListDoesNotFallBack(t *testing.T) {
885 bOrigin, bHits := hitServer(t)
886 var calls atomic.Int64
887 var pl pollLog
888 list := scripted(&calls,
889 sweep{recs: []apiInstanceRecord{rec("c.test", nil)}, truncated: true},
890 sweep{recs: []apiInstanceRecord{
891 rec("b.test", &apiRollout{Box: "x1", Installed: B, Good: "b1", SchedulerIdle: true}),
892 rec("c.test", nil),
893 }},
894 )
895 rw := testWatch(io.Discard, list, pl.check(map[string]bool{"c.test": true}))
896 targets := []watchTarget{{Host: "b.test", Origin: bOrigin}, {Host: "c.test", Origin: "http://127.0.0.1:1"}}
897 if err := rw.run(targets); err != nil {
898 t.Fatal(err)
899 }
900 if got := pl.hosts(); bHits.Load() != 0 || got["b.test"] != 0 || got["c.test"] != 1 {
901 t.Fatalf("b.test origin hit %d times, check polled %v", bHits.Load(), got)
902 }
903}
904
905// Fix round 1 (R6): a code this binary has no phrase for is a newer box's
906// problem; its instances are followed and never requested.
907func TestWatchRolloutUnknownUnavailableCodeIsNeverPolled(t *testing.T) {
908 origin, hits := hitServer(t)
909 var calls atomic.Int64
910 list := sweeps(&calls, []apiInstanceRecord{rec("a.test", &apiRollout{Box: "x1", Installed: B, Unavailable: "future-code"})})
911 err := testWatch(io.Discard, list, noURL(t)).run([]watchTarget{{Host: "a.test", Origin: origin}})
912 if hits.Load() != 0 || err == nil || !strings.Contains(err.Error(), "a.test: "+rolloutwords.Unavailable("future-code")) {
913 t.Fatalf("%d HTTP polls, err = %v", hits.Load(), err)
914 }
915}
916
917// Astra plan 6 R3: a target missing from a complete, untruncated list is not
918// evidence that it uses the old path (the console may simply not have been
919// able to read its record). It is never polled, and the timeout names it.
920func TestWatchRolloutNeverPollsATargetMissingFromACompleteList(t *testing.T) {
921 var calls atomic.Int64
922 settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
923 list := scripted(&calls, sweep{recs: []apiInstanceRecord{rec("a.test", settling)}})
924 rw := testWatch(io.Discard, list, noURL(t))
925 onURL := 0
926 rw.onURL = func() { onURL++ }
927 err := rw.run(targetsFor("a.test", "b.test"))
928 if onURL != 0 {
929 t.Fatalf("onURL ran %d times: a missing host was watched by URL", onURL)
930 }
931 if err == nil || !strings.Contains(err.Error(), "\n b.test: "+watchPendingUnread) {
932 t.Fatalf("err = %v, want a timeout naming b.test", err)
933 }
934}
935
936// A host first listed with no report on a complete join is URL-watched. If it
937// then drops off the list (its record became unreadable), the watch must stop
938// requesting it: the request would wake it if it is asleep.
939func TestWatchRolloutStopsPollingAURLHostThatDropsOffTheList(t *testing.T) {
940 var calls, polls, atDrop atomic.Int64
941 settling := &apiRollout{Box: "x1", Installed: B, Serving: "b1", Awake: true, Attempt: status.AttemptSettling, AttemptBuild: B}
942 list := func(time.Time) (watchList, error) {
943 if calls.Add(1) == 1 {
944 return watchList{recs: []apiInstanceRecord{rec("a.test", settling), rec("b.test", nil)}}, nil
945 }
946 atDrop.CompareAndSwap(0, polls.Load()+1) // +1 so a snapshot of zero polls is not mistaken for unset
947 return watchList{recs: []apiInstanceRecord{rec("a.test", settling)}}, nil
948 }
949 check := func(tg watchTarget, _ int64) (bool, string) {
950 if tg.Host != "b.test" {
951 t.Errorf("%s was polled over HTTP", tg.Host)
952 }
953 polls.Add(1)
954 return false, "not yet"
955 }
956 err := testWatch(io.Discard, list, check).run(targetsFor("a.test", "b.test"))
957 if err == nil {
958 t.Fatal("the watch succeeded; a was never final")
959 }
960 if atDrop.Load() < 2 {
961 t.Fatalf("b was never polled before it dropped off the list (snapshot %d)", atDrop.Load())
962 }
963 if got, want := polls.Load(), atDrop.Load()-1; got != want {
964 t.Fatalf("%d polls, %d when b dropped off the list: b was polled after it was gone", got, want)
965 }
966}
967