| 1 | package rollout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "slices" |
| 6 | "strings" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/activator" |
| 11 | "github.com/carlosframework/platform/internal/appconfig" |
| 12 | ) |
| 13 | |
| 14 | // restartOn is an agent restart under the harness: a fresh Manager over |
| 15 | // book, its counters (attempt numbers, wake tokens) starting again from |
| 16 | // zero, as a new process's do. The old manager's schedulers stop with ctx. |
| 17 | func (h *harness) restartOn(book *FailureBook) { |
| 18 | h.t.Helper() |
| 19 | book.now = h.clock |
| 20 | h.book = book |
| 21 | ctx, cancel := context.WithCancel(context.Background()) |
| 22 | h.t.Cleanup(cancel) |
| 23 | h.m = NewManager(ctx, Deps{ |
| 24 | Act: h.probe, Routes: h.rows, Book: book, OptDir: h.opt, DataDir: h.data, Box: "box-1", |
| 25 | Demand: h.demand, Policy: h.policy, Pointer: h.pointerFor, Events: h.events, |
| 26 | Touch: func() { h.touched.Add(1) }, Now: h.clock, Logf: func(string, ...any) {}, |
| 27 | }) |
| 28 | } |
| 29 | |
| 30 | // cancelHoldWake plans host's first wake (a hold build) and ends its |
| 31 | // observation cancelled after the build ran, which leaves the host's |
| 32 | // hold-ran marker and no failure record. |
| 33 | func (h *harness) cancelHoldWake(host, build string) activator.WakePlan { |
| 34 | h.t.Helper() |
| 35 | h.wakeHost(host) |
| 36 | p := h.m.Plan(context.Background(), h.rows.get(host)) |
| 37 | if !p.FirstWake || p.Boot.Version != build || p.Mode != activator.ModeHold { |
| 38 | h.t.Fatalf("plan %+v: want a hold first wake of %s", p, build) |
| 39 | } |
| 40 | g := activator.Gen{N: 7, Build: build, StagedPath: p.Boot.StagedPath, Digest: p.Boot.Digest} |
| 41 | h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true, Token: p.Token}) |
| 42 | h.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.rows.get(host), Gen: g, Mode: p.Mode, FirstWake: true, |
| 43 | Token: p.Token, Executed: true}) |
| 44 | return p |
| 45 | } |
| 46 | |
| 47 | // Final review round 1, finding 1 (Astra): execution identities are unique |
| 48 | // across agent restarts. B runs on h and leaves a marker with no failure |
| 49 | // record; the agent restarts (a new Manager, its counters at zero again, |
| 50 | // over the same book on disk) and its first wake runs B again. The second |
| 51 | // execution is a new one: the marker gets a fresh Seq and At, so the retry |
| 52 | // grant the console signed for the first execution, and a rollback pointer |
| 53 | // signed between the two executions, clear nothing of it. |
| 54 | func TestAnExecutionAfterARestartIsANewExecution(t *testing.T) { |
| 55 | dir := t.TempDir() |
| 56 | h := newHarness(t, "s") |
| 57 | book, err := OpenFailureBook(dir) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | h.restartOn(book) |
| 62 | h.swap = appconfig.SwapHold |
| 63 | h.stage("v2") |
| 64 | h.install("v2", "s") |
| 65 | h.sleepHost("s") |
| 66 | h.pass() |
| 67 | h.cancelHoldWake("s", "v2") |
| 68 | first, ok := h.book.HoldRanOn("s") |
| 69 | if !ok || first.Build != "v2" || !first.At.Equal(t0) { |
| 70 | t.Fatalf("first marker %+v, %v: want v2's at %v", first, ok, t0) |
| 71 | } |
| 72 | staleGrant := h.book.RestrictionEpoch("bab", "blog", "v2", "s") |
| 73 | |
| 74 | // The agent restarts: the book reopens from disk, the manager's |
| 75 | // counters start again. |
| 76 | reopened, err := OpenFailureBook(dir) |
| 77 | if err != nil { |
| 78 | t.Fatal(err) |
| 79 | } |
| 80 | h.restartOn(reopened) |
| 81 | h.advance(time.Minute) |
| 82 | second := h.clock() |
| 83 | h.pass() |
| 84 | h.cancelHoldWake("s", "v2") |
| 85 | got, ok := h.book.HoldRanOn("s") |
| 86 | if !ok || got.Build != "v2" { |
| 87 | t.Fatalf("marker after the second execution %+v, %v", got, ok) |
| 88 | } |
| 89 | if got.Exec == first.Exec { |
| 90 | t.Fatalf("the second execution reused the first's identity %q", got.Exec) |
| 91 | } |
| 92 | if got.Seq <= first.Seq || !got.At.Equal(second) { |
| 93 | t.Fatalf("marker %+v after the second execution: want a fresh Seq past %d and At %v", got, first.Seq, second) |
| 94 | } |
| 95 | |
| 96 | // The first execution's grant and a rollback signed between the two runs |
| 97 | // clear nothing of the second's marker. |
| 98 | if rec, mark, _, _ := h.book.RetryHostAt("bab", "blog", "v2", "s", h.book.ID(), staleGrant); rec || mark { |
| 99 | t.Fatal("a retry grant signed for the first execution cleared the second's marker") |
| 100 | } |
| 101 | if ok, _ := h.book.ClearHoldRanOnPointer("s", "bab", "blog", t0.Add(30*time.Second)); ok { |
| 102 | t.Fatal("a rollback signed between the two executions cleared the second's marker") |
| 103 | } |
| 104 | if _, ok := h.book.HoldRanOn("s"); !ok { |
| 105 | t.Fatal("the second execution's marker is gone") |
| 106 | } |
| 107 | } |
| 108 | |
| 109 | // Finding 1's identities: an attempt's and a wake's carry the manager's |
| 110 | // incarnation, so two managers (two agent processes) never hand out the same |
| 111 | // one, and the a/w prefix Begin's wake rule reads is kept. |
| 112 | func TestExecutionIdentitiesCarryTheManagersIncarnation(t *testing.T) { |
| 113 | m1 := NewManager(context.Background(), Deps{}) |
| 114 | m2 := NewManager(context.Background(), Deps{}) |
| 115 | a1, a2 := m1.nextExecID(), m2.nextExecID() |
| 116 | if a1 == a2 || !strings.HasPrefix(a1, "a") || !strings.HasPrefix(a2, "a") || wakeIntent(a1) { |
| 117 | t.Fatalf("attempt identities %q and %q: want distinct, both attempts", a1, a2) |
| 118 | } |
| 119 | w1, w2 := m1.wakeExecID(1), m2.wakeExecID(1) |
| 120 | if w1 == w2 || !wakeIntent(w1) || !wakeIntent(w2) { |
| 121 | t.Fatalf("wake identities %q and %q: want distinct, both wakes", w1, w2) |
| 122 | } |
| 123 | if m1.wakeExecID(0) != "" { |
| 124 | t.Fatal("token zero has an identity") |
| 125 | } |
| 126 | if again := m1.nextExecID(); again == a1 { |
| 127 | t.Fatalf("the same manager handed out %q twice", a1) |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | // grant is a console's host retry grant as the box's status lets it sign |
| 132 | // one: the restriction's book and epoch. |
| 133 | type grant struct { |
| 134 | book string |
| 135 | epoch uint64 |
| 136 | } |
| 137 | |
| 138 | func grantFor(b *FailureBook, build, host string) grant { |
| 139 | return grant{book: b.ID(), epoch: b.RestrictionEpoch("bab", "blog", build, host)} |
| 140 | } |
| 141 | |
| 142 | func applyGrant(b *FailureBook, build, host string, g grant) (rec, mark bool) { |
| 143 | rec, mark, _, _ = b.RetryHostAt("bab", "blog", build, host, g.book, g.epoch) |
| 144 | return rec, mark |
| 145 | } |
| 146 | |
| 147 | // Final review round 1, finding 3 (Astra): two independent books (a host |
| 148 | // relocated, or a box rebuilt) count from the same place, so the same |
| 149 | // numeric epoch names different restrictions in each. A grant signed for |
| 150 | // one book's restriction clears nothing in the other's. |
| 151 | func TestARetryGrantForOneBookClearsNothingInAnother(t *testing.T) { |
| 152 | one, other := &FailureBook{}, &FailureBook{} |
| 153 | for _, b := range []*FailureBook{one, other} { |
| 154 | b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Mode: string(activator.ModeHold), |
| 155 | HeldDown: []string{"h"}, MarkHoldRan: true, HoldRanExec: "a1", HoldRanAt: t0}) |
| 156 | } |
| 157 | g := grantFor(one, "v2", "h") |
| 158 | if g.epoch == 0 || g.epoch != other.RestrictionEpoch("bab", "blog", "v2", "h") { |
| 159 | t.Fatalf("the two books' epochs differ (%d, %d): the case needs equal counters", g.epoch, other.RestrictionEpoch("bab", "blog", "v2", "h")) |
| 160 | } |
| 161 | if rec, mark := applyGrant(other, "v2", "h", g); rec || mark { |
| 162 | t.Fatalf("a grant for one book cleared record %v, marker %v in another", rec, mark) |
| 163 | } |
| 164 | if !other.HeldDown("bab", "blog", "v2", "h") { |
| 165 | t.Fatal("the other book's hold-down was lifted") |
| 166 | } |
| 167 | if rec, mark := applyGrant(one, "v2", "h", g); !rec || !mark { |
| 168 | t.Fatalf("the grant cleared record %v, marker %v in its own book; want both", rec, mark) |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // Finding 3's failed-save half, under the crash-epoch hard limit (readiness |
| 173 | // round 2, Task 1, which replaced the reopen's bounded advance): a |
| 174 | // generation handed out after a save failed lives only in memory, so status |
| 175 | // never publishes it and no grant at it clears anything, before a crash or |
| 176 | // after. A reopen may hand the same number out again, which no grant can |
| 177 | // name; every value it hands out is past every one status published. |
| 178 | func TestAnUnsavedGenerationIsNeverPublished(t *testing.T) { |
| 179 | dd := t.TempDir() |
| 180 | b, err := OpenFailureBook(dd) |
| 181 | if err != nil { |
| 182 | t.Fatal(err) |
| 183 | } |
| 184 | if err := b.Flush(); err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | if _, err := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h"}); err != nil { |
| 188 | t.Fatal(err) |
| 189 | } |
| 190 | saved := b.PublishedEpoch("bab", "blog", "v2", "h") |
| 191 | if saved == 0 { |
| 192 | t.Fatal("a saved restriction published nothing") |
| 193 | } |
| 194 | readOnly(t, dd) |
| 195 | if _, err := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "h", HeldDown: []string{"h"}}); err == nil { |
| 196 | t.Skip("the save did not fail") |
| 197 | } |
| 198 | if e := b.PublishedEpoch("bab", "blog", "v3", "h"); e != 0 { |
| 199 | t.Fatalf("status published %d for a restriction not on disk, want 0", e) |
| 200 | } |
| 201 | unsaved := grantFor(b, "v3", "h") // a console that signed the in-memory value anyway |
| 202 | if rec, mark := applyGrant(b, "v3", "h", unsaved); rec || mark { |
| 203 | t.Fatal("a grant at the unsaved generation cleared before the save landed") |
| 204 | } |
| 205 | writable(t, dd) |
| 206 | |
| 207 | b2, err := OpenFailureBook(dd) // the agent crashed: v3's record never reached the disk |
| 208 | if err != nil { |
| 209 | t.Fatal(err) |
| 210 | } |
| 211 | if err := b2.Flush(); err != nil { |
| 212 | t.Fatal(err) |
| 213 | } |
| 214 | b2.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "h", HeldDown: []string{"h"}}) |
| 215 | if e := b2.PublishedEpoch("bab", "blog", "v3", "h"); e <= saved { |
| 216 | t.Fatalf("the reopened book published %d, not past the published %d", e, saved) |
| 217 | } |
| 218 | if rec, mark := applyGrant(b2, "v2", "h", grant{book: b.ID(), epoch: saved}); rec || mark { |
| 219 | t.Fatal("the grant for the published restriction cleared after the app's restrictions changed") |
| 220 | } |
| 221 | } |
| 222 | |
| 223 | // Final review round 1, finding 4 (Astra): a retry that removes only a |
| 224 | // hold-ran marker (no failure record holds the host) is still a retry. An |
| 225 | // execution of the marked build that began before it, a scheduled attempt or |
| 226 | // a planned wake, and reports failure after it, is stale: no record is |
| 227 | // written and the cohort is not halted. A no-op retry changes nothing |
| 228 | // (TestANoOpRetryNotesNoClear). |
| 229 | func TestAMarkerOnlyRetryOrdersALateFailureBeforeIt(t *testing.T) { |
| 230 | for _, kind := range []string{"attempt", "wake"} { |
| 231 | t.Run(kind, func(t *testing.T) { |
| 232 | ctx := context.Background() |
| 233 | k := appKey{"bab", "blog"} |
| 234 | h := newHarness(t, "h") |
| 235 | h.swap = appconfig.SwapHold |
| 236 | h.stage("v2") |
| 237 | h.book.MarkHoldRan("h", HoldRan{Account: "bab", App: "blog", Build: "v2", Exec: "a-earlier", At: t0}) |
| 238 | var plan activator.WakePlan |
| 239 | if kind == "attempt" { |
| 240 | gate := h.act.hold("h") |
| 241 | h.install("v2", "h") |
| 242 | h.pass() |
| 243 | waitFor(t, "h's attempt onto v2", func() bool { |
| 244 | _, started, _ := h.act.snapshot() |
| 245 | return len(started) == 1 && started[0] == "h:v2" |
| 246 | }) |
| 247 | if !h.retry("h", "v2") { |
| 248 | close(gate) |
| 249 | t.Fatal("the marker-only retry cleared nothing") |
| 250 | } |
| 251 | h.act.mu.Lock() |
| 252 | h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapFailed } |
| 253 | h.act.mu.Unlock() |
| 254 | next := h.act.hold("h") // a next attempt, if any, parks in Swap |
| 255 | defer close(next) |
| 256 | close(gate) |
| 257 | waitFor(t, "h's attempt to finish", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 }) |
| 258 | } else { |
| 259 | h.install("v2", "h") |
| 260 | h.sleepHost("h") |
| 261 | h.pass() |
| 262 | h.wakeHost("h") |
| 263 | plan = h.m.Plan(ctx, h.rows.get("h")) |
| 264 | if !plan.FirstWake || plan.Boot.Version != "v2" { |
| 265 | t.Fatalf("plan %+v: want a first wake of the marked v2", plan) |
| 266 | } |
| 267 | if !h.retry("h", "v2") { |
| 268 | t.Fatal("the marker-only retry cleared nothing") |
| 269 | } |
| 270 | h.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("h"), Failed: plan.Boot, Mode: plan.Mode, |
| 271 | FirstWake: true, Reason: "health check failed", Token: plan.Token, ConfigRev: plan.ConfigRev}) |
| 272 | } |
| 273 | if _, ok := h.book.Lookup("bab", "blog", "v2"); ok { |
| 274 | t.Fatal("a failure of an execution that began before the retry wrote a record") |
| 275 | } |
| 276 | s := h.m.scheduler(k) |
| 277 | s.mu.Lock() |
| 278 | halted := s.q.cohortFor(cohortKey("v2", classAdopt)).halted |
| 279 | s.mu.Unlock() |
| 280 | if halted { |
| 281 | t.Fatal("a failure of an execution that began before the retry halted the cohort") |
| 282 | } |
| 283 | }) |
| 284 | } |
| 285 | } |
| 286 | |
| 287 | // Finding 4's other half: a retry that clears nothing (no record, no marker |
| 288 | // of the build) notes no clear, so a failure of an execution that began |
| 289 | // before it is still recorded. |
| 290 | func TestANoOpRetryNotesNoClear(t *testing.T) { |
| 291 | b := &FailureBook{} |
| 292 | snap := b.Snapshot() |
| 293 | if rec, mark, _, _ := b.RetryHostAt("bab", "blog", "v2", "h", b.ID(), 1); rec || mark { |
| 294 | t.Fatal("a retry with nothing to clear cleared something") |
| 295 | } |
| 296 | if res, _ := b.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "h", Seq: snap}); res != Recorded { |
| 297 | t.Fatalf("a failure after a no-op retry = %v, want recorded", res) |
| 298 | } |
| 299 | } |
| 300 | |
| 301 | // Final review round 1, finding 5 (Astra): a host retry for a build B the |
| 302 | // slot does not want now (it wants A) still entitles the host to a canary |
| 303 | // onto B. The entitlement survives an attempt onto another build in |
| 304 | // between, and is spent only by the adoption onto B: when the slot comes |
| 305 | // to want B while B's cohort is halted, that adoption is admitted as the |
| 306 | // host's canary. |
| 307 | func TestARetryForABuildNotWantedYetKeepsItsCanary(t *testing.T) { |
| 308 | k := appKey{"bab", "blog"} |
| 309 | h := newHarness(t, "h", "s") |
| 310 | h.stage("v2", "v3") |
| 311 | h.act.mu.Lock() |
| 312 | h.act.outcome = func(a activator.Attempt) activator.SwapOutcome { |
| 313 | if a.Host == "s" && a.Build == "v3" { |
| 314 | return activator.SwapFailed |
| 315 | } |
| 316 | return activator.SwapDone |
| 317 | } |
| 318 | h.act.mu.Unlock() |
| 319 | h.pass() |
| 320 | |
| 321 | // s fails v3: a record holds v3 for h too, and v3's cohort halts. |
| 322 | h.install("v3", "s") |
| 323 | h.pass() |
| 324 | waitFor(t, "s's attempt onto v3 to fail", func() bool { calls, _, _ := h.act.snapshot(); return len(calls) == 1 }) |
| 325 | s := h.m.scheduler(k) |
| 326 | halted := func() bool { |
| 327 | s.mu.Lock() |
| 328 | defer s.mu.Unlock() |
| 329 | return s.q.cohortFor(cohortKey("v3", classAdopt)).halted |
| 330 | } |
| 331 | waitFor(t, "v3's cohort to halt", halted) |
| 332 | |
| 333 | // The console's retry of h names v3; h's slot wants v1, which it serves. |
| 334 | if !h.retry("h", "v3") { |
| 335 | t.Fatal("the retry for v3 cleared nothing") |
| 336 | } |
| 337 | // An adoption onto another build (A = v2) runs in between: it does not |
| 338 | // spend the entitlement for v3. |
| 339 | h.install("v2", "h") |
| 340 | h.pass() |
| 341 | waitFor(t, "h's attempt onto v2", func() bool { return h.servingAll("v2", "h") && !h.m.attemptRunning("h") }) |
| 342 | |
| 343 | // Now h's slot wants v3, its cohort still halted: the adoption onto v3 is |
| 344 | // admitted, as h's canary. |
| 345 | if !halted() { |
| 346 | t.Fatal("v3's cohort is no longer halted: the case needs the halt") |
| 347 | } |
| 348 | gate := h.act.hold("h") |
| 349 | defer close(gate) |
| 350 | h.install("v3", "h") |
| 351 | h.pass() |
| 352 | waitFor(t, "h's adoption onto v3 through the halt", func() bool { |
| 353 | _, started, _ := h.act.snapshot() |
| 354 | return slices.Contains(started, "h:v3") |
| 355 | }) |
| 356 | s.mu.Lock() |
| 357 | d := s.q.hosts["h"].running |
| 358 | s.mu.Unlock() |
| 359 | if d == nil || d.ID.Build != "v3" || !d.Canary || !d.Forced { |
| 360 | t.Fatalf("h's running attempt %+v: want the forced canary onto v3", d) |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | // The entitlement belongs to its build: it never lets an adoption onto |
| 365 | // another build through that build's halt. |
| 366 | func TestARetryEntitlementIsNeverSpentOnAnotherBuild(t *testing.T) { |
| 367 | q := newAppQueue("bab", "blog", func(string, string, bool) bool { return false }) |
| 368 | q.concurrency = 1 |
| 369 | now := t0 |
| 370 | q.observe("h", "v1", "v1", now) |
| 371 | q.wantBuild("h", activator.CauseRelease, "v1", now) |
| 372 | q.retryHost("h", "v3") |
| 373 | q.cohortFor(cohortKey("v2", classAdopt)).halted = true |
| 374 | q.wantBuild("h", activator.CauseRelease, "v2", now) |
| 375 | if _, ok := q.gate(q.hosts["h"], now); ok { |
| 376 | t.Fatal("an entitlement for v3 let an adoption onto v2 through v2's halt") |
| 377 | } |
| 378 | q.cohortFor(cohortKey("v3", classAdopt)).halted = true |
| 379 | q.wantBuild("h", activator.CauseRelease, "v3", now) |
| 380 | d, ok := q.gate(q.hosts["h"], now) |
| 381 | if !ok || !d.Forced || !d.Canary { |
| 382 | t.Fatalf("gate onto v3 = %+v, %v: want the forced canary", d, ok) |
| 383 | } |
| 384 | } |
| 385 | |
amadan