| 1 | package rollout |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "testing" |
| 6 | "time" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/activator" |
| 9 | "github.com/carlosframework/platform/internal/release" |
| 10 | ) |
| 11 | |
| 12 | // workerHarness turns the harness's hosts into always-on sidecars. |
| 13 | func workerHarness(t *testing.T, hosts ...string) *harness { |
| 14 | h := newHarness(t, hosts...) |
| 15 | for _, host := range hosts { |
| 16 | r := h.rows.get(host) |
| 17 | r.Kind, r.Hibernate = "sidecar", false |
| 18 | h.rows.put(r) |
| 19 | } |
| 20 | return h |
| 21 | } |
| 22 | |
| 23 | // Spec §7.9: a socketless attempt is ReplaceWorker on the same entry, never |
| 24 | // Swap and never a stop and a wake. It carries the host's good build as its |
| 25 | // fallback. |
| 26 | func TestAWorkerReleaseReplacesOnTheSameEntry(t *testing.T) { |
| 27 | h := workerHarness(t, "bot") |
| 28 | h.stage("v2") |
| 29 | h.install("v2", "bot") |
| 30 | h.pass() |
| 31 | waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" }) |
| 32 | if calls, _, _ := h.act.snapshot(); len(calls) != 0 { |
| 33 | t.Fatal("a socketless attempt went through Swap") |
| 34 | } |
| 35 | r := h.act.replacements() |
| 36 | if len(r) != 1 || r[0].a.Build != "v2" || r[0].fallback == nil || r[0].fallback.Version != "v1" { |
| 37 | t.Fatalf("replacements %+v, want one onto v2 with the good v1 behind it", r) |
| 38 | } |
| 39 | if fb := r[0].fallback; fb.Digest != "sha-v1" || fb.StagedPath != release.VersionPath(h.opt, "bab", "v1", "blog") { |
| 40 | t.Fatalf("fallback %+v, want the good build by its retained identity", fb) |
| 41 | } |
| 42 | } |
| 43 | |
| 44 | // Spec §6.5, §8.5 (Astra R3 flag triage T9): a replacement whose build exits |
| 45 | // cleanly as a one-shot leaves nothing for the activator to serve, so the |
| 46 | // acknowledged check would drop its good write. done() records the build |
| 47 | // good from the result, and clears a hold-ran marker older than the attempt. |
| 48 | // A replacement that is not a clean exit still needs the activator to serve |
| 49 | // it. |
| 50 | func TestACleanOneShotReplacementRecordsItsBuildGood(t *testing.T) { |
| 51 | h := workerHarness(t, "bot") |
| 52 | h.stage("v2") |
| 53 | h.book.MarkHoldRan("bot", HoldRan{Account: "bab", App: "blog", Build: "v9"}) // set before the attempt starts |
| 54 | h.act.mu.Lock() |
| 55 | h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult { |
| 56 | return activator.SwapResult{Outcome: activator.SwapDone, CleanExit: true} |
| 57 | } |
| 58 | h.act.mu.Unlock() |
| 59 | h.install("v2", "bot") |
| 60 | h.pass() |
| 61 | waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" }) |
| 62 | if r := h.rows.get("bot"); r.GoodDigest != "sha-v2" { |
| 63 | t.Fatalf("row = %+v, want v2's digest as the good build's", r) |
| 64 | } |
| 65 | if _, ok := h.act.Current("bot"); ok { |
| 66 | t.Fatal("the fake still serves the one-shot: the test does not model a parked worker") |
| 67 | } |
| 68 | if _, held := h.book.HoldRanOn("bot"); held { |
| 69 | t.Fatal("the clean one-shot passed, but the older hold-ran marker stands") |
| 70 | } |
| 71 | } |
| 72 | |
| 73 | // Astra R3 N5: a hold build that ran on a worker without passing (a |
| 74 | // superseded hold, its marker set by done()) may have migrated its |
| 75 | // database, so the next scheduled replacement carries no fallback onto the |
| 76 | // good build. Another app's marker under the same host name changes nothing. |
| 77 | func TestAWorkerReplacementCarriesNoFallbackWhileAMarkerStands(t *testing.T) { |
| 78 | h := workerHarness(t, "bot") |
| 79 | h.stage("v2", "v3") |
| 80 | h.book.MarkHoldRan("bot", markV2) |
| 81 | h.install("v3", "bot") |
| 82 | h.pass() |
| 83 | waitFor(t, "the replacement onto v3", func() bool { return len(h.act.replacements()) == 1 }) |
| 84 | if r := h.act.replacements(); r[0].a.Build != "v3" || r[0].fallback != nil { |
| 85 | t.Fatalf("replacement %+v (fallback %+v): want v3 with no fallback while v2's marker stands", r[0].a, r[0].fallback) |
| 86 | } |
| 87 | |
| 88 | other := workerHarness(t, "bot") |
| 89 | other.stage("v3") |
| 90 | other.book.MarkHoldRan("bot", HoldRan{Account: "bab", App: "other", Build: "v2"}) |
| 91 | other.install("v3", "bot") |
| 92 | other.pass() |
| 93 | waitFor(t, "the replacement onto v3", func() bool { return len(other.act.replacements()) == 1 }) |
| 94 | if r := other.act.replacements(); r[0].fallback == nil || r[0].fallback.Version != "v1" { |
| 95 | t.Fatalf("replacement %+v: another app's marker took away the good v1 fallback", r[0]) |
| 96 | } |
| 97 | } |
| 98 | |
| 99 | func TestAFailedWorkerReleaseFallsBackAndHalts(t *testing.T) { |
| 100 | h := workerHarness(t, "bot") |
| 101 | h.stage("v2") |
| 102 | h.act.mu.Lock() |
| 103 | h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult { |
| 104 | return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"} |
| 105 | } |
| 106 | h.act.mu.Unlock() |
| 107 | h.install("v2", "bot") |
| 108 | h.pass() |
| 109 | waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok }) |
| 110 | if r := h.rows.get("bot"); r.GoodVersion != "v1" { |
| 111 | t.Fatalf("good build = %q after a failed worker release", r.GoodVersion) |
| 112 | } |
| 113 | } |
| 114 | |
| 115 | // Spec §8.1 (P5 review finding 7): the fallback's held publication is |
| 116 | // released only after done() has recorded the failure it falls back from. |
| 117 | func TestAWorkerFallbackIsPublishedAfterItsFailureIsRecorded(t *testing.T) { |
| 118 | h := workerHarness(t, "bot") |
| 119 | h.stage("v2") |
| 120 | recorded := make(chan bool, 1) |
| 121 | h.act.mu.Lock() |
| 122 | h.act.worker = func(a activator.Attempt, fb *release.StagedIdentity) activator.SwapResult { |
| 123 | return activator.SwapResult{Outcome: activator.SwapFailed, Reason: "exited within 10s of starting"} |
| 124 | } |
| 125 | h.act.onHeldPublish = func(string) { |
| 126 | _, ok := h.book.Lookup("bab", "blog", "v2") |
| 127 | select { |
| 128 | case recorded <- ok: |
| 129 | default: |
| 130 | } |
| 131 | } |
| 132 | h.act.mu.Unlock() |
| 133 | h.install("v2", "bot") |
| 134 | h.pass() |
| 135 | select { |
| 136 | case ok := <-recorded: |
| 137 | if !ok { |
| 138 | t.Fatal("the fallback's publication was released before v2's failure was recorded") |
| 139 | } |
| 140 | case <-time.After(5 * time.Second): |
| 141 | t.Fatal("the held publication was never released") |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | // Spec §7.9: a recycle that falls due coalesces with a queued release attempt |
| 146 | // only when the scheduler would admit that attempt now. |
| 147 | func TestARecycleCoalescesOnlyWithAnAdmissibleRelease(t *testing.T) { |
| 148 | q := awakeQueue(noBlocks, "a", "b") |
| 149 | releaseAll(q, "v2", "a", "b") |
| 150 | canary, ok := q.admitOne("a", t0) |
| 151 | if !ok || !canary.Canary { |
| 152 | t.Fatalf("admitOne(a) = %+v %v, want a's canary", canary, ok) |
| 153 | } |
| 154 | if _, ok := q.admitOne("b", t0); ok { |
| 155 | t.Fatal("coalesced past swap_concurrency 1 while another worker's canary runs") |
| 156 | } |
| 157 | q.finish(canary, activator.SwapDone) |
| 158 | if d, ok := q.admitOne("b", t0); !ok || d.ID.Build != "v2" { |
| 159 | t.Fatalf("admitOne(b) = %+v %v after the canary settled", d, ok) |
| 160 | } |
| 161 | r := awakeQueue(noBlocks, "c") |
| 162 | r.wantRestart("c", "r1", t0) |
| 163 | if _, ok := r.admitOne("c", t0); ok { |
| 164 | t.Fatal("a restart is not a release attempt; the recycle runs on the serving build") |
| 165 | } |
| 166 | h := newHarness(t, "a") |
| 167 | if h.m.CoalesceRecycle("nobody") { |
| 168 | t.Fatal("an unknown host coalesced") |
| 169 | } |
| 170 | } |
| 171 | |
| 172 | // Spec §7.9: while an attempt runs for the host, the recycle is skipped: |
| 173 | // that attempt's ReplaceWorker makes the spawn. A crash during its |
| 174 | // observation is the build's to report, never a recycle's to paper over. |
| 175 | func TestARecycleIsSkippedWhileTheHostsAttemptRuns(t *testing.T) { |
| 176 | h := workerHarness(t, "bot") |
| 177 | h.stage("v2") |
| 178 | running := make(chan struct{}) |
| 179 | proceed := make(chan struct{}) |
| 180 | h.act.mu.Lock() |
| 181 | h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult { |
| 182 | close(running) |
| 183 | <-proceed |
| 184 | return activator.SwapResult{Outcome: activator.SwapDone} |
| 185 | } |
| 186 | h.act.mu.Unlock() |
| 187 | h.install("v2", "bot") |
| 188 | h.pass() |
| 189 | select { |
| 190 | case <-running: |
| 191 | case <-time.After(5 * time.Second): |
| 192 | t.Fatal("the attempt never started") |
| 193 | } |
| 194 | if !h.m.CoalesceRecycle("bot") { |
| 195 | t.Fatal("a recycle went ahead while the host's attempt was running") |
| 196 | } |
| 197 | close(proceed) |
| 198 | waitFor(t, "the attempt to finish", func() bool { return h.rows.get("bot").GoodVersion == "v2" }) |
| 199 | if h.m.CoalesceRecycle("bot") { |
| 200 | t.Fatal("coalesced with nothing queued or running") |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // Plan 4's contract: CoalesceRecycle is called under the activator's lock, |
| 205 | // so it must never call back into the activator. Here the fake's own lock |
| 206 | // stands in for activator.Service.mu, held across the call: a callback into |
| 207 | // Current, Owns or ReplaceWorker would never return. |
| 208 | func TestCoalesceRecycleNeverCallsTheActivator(t *testing.T) { |
| 209 | h := workerHarness(t, "bot") |
| 210 | h.pass() // the host is known to its scheduler |
| 211 | waitFor(t, "the host's scheduler", func() bool { return h.m.schedulerOf("bot") != nil }) |
| 212 | answered := make(chan bool, 1) |
| 213 | h.act.mu.Lock() |
| 214 | go func() { answered <- h.m.CoalesceRecycle("bot") }() |
| 215 | select { |
| 216 | case <-answered: |
| 217 | case <-time.After(2 * time.Second): |
| 218 | h.act.mu.Unlock() |
| 219 | t.Fatal("CoalesceRecycle waited on the activator") |
| 220 | } |
| 221 | h.act.mu.Unlock() |
| 222 | } |
| 223 | |
| 224 | // Task 11 fix round 1, review finding 2 (spec §8.6): an adoption a recycle |
| 225 | // coalesces with runs on its own goroutine, and Close joins it as it joins |
| 226 | // one the scheduler loop admits: its outcome and its End are on disk when |
| 227 | // Close returns. Once Close has begun a recycle admits nothing; it is |
| 228 | // skipped, which is all a stopping agent needs. |
| 229 | func TestCloseJoinsACoalescedWorkerAttempt(t *testing.T) { |
| 230 | h := workerHarness(t, "bot") |
| 231 | dir := h.onDisk() |
| 232 | h.stage("v2", "v3") |
| 233 | h.pass() |
| 234 | waitFor(t, "the host's scheduler to see it serve v1", func() bool { |
| 235 | s := h.m.schedulerOf("bot") |
| 236 | if s == nil { |
| 237 | return false |
| 238 | } |
| 239 | s.mu.Lock() |
| 240 | defer s.mu.Unlock() |
| 241 | hs := s.q.hosts["bot"] |
| 242 | return hs != nil && hs.serving == "v1" |
| 243 | }) |
| 244 | running, proceed := make(chan struct{}), make(chan struct{}) |
| 245 | h.act.mu.Lock() |
| 246 | h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult { |
| 247 | close(running) |
| 248 | <-proceed |
| 249 | return activator.SwapResult{Outcome: activator.SwapDone} |
| 250 | } |
| 251 | h.act.mu.Unlock() |
| 252 | queue := func(build string) { |
| 253 | h.install(build, "bot") |
| 254 | s := h.m.schedulerOf("bot") |
| 255 | s.mu.Lock() |
| 256 | s.q.wantBuild("bot", activator.CauseRelease, build, h.clock()) // queued, not yet admitted by the loop |
| 257 | s.mu.Unlock() |
| 258 | } |
| 259 | queue("v2") |
| 260 | if !h.m.CoalesceRecycle("bot") { |
| 261 | t.Fatal("the recycle did not coalesce with the admissible release") |
| 262 | } |
| 263 | select { |
| 264 | case <-running: |
| 265 | case <-time.After(5 * time.Second): |
| 266 | t.Fatal("the coalesced attempt never started") |
| 267 | } |
| 268 | if n := len(fileOf(t, dir).Intents); n != 1 { |
| 269 | t.Fatalf("%d intents on disk while the coalesced attempt runs, want 1", n) |
| 270 | } |
| 271 | go func() { time.Sleep(200 * time.Millisecond); close(proceed) }() |
| 272 | ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| 273 | defer cancel() |
| 274 | h.m.Close(ctx) |
| 275 | if f := fileOf(t, dir); len(f.Intents) != 0 { |
| 276 | t.Fatalf("intents %+v on disk after Close: a restart would reconcile the coalesced attempt as a failure", f.Intents) |
| 277 | } |
| 278 | if g := h.rows.get("bot").GoodVersion; g != "v2" { |
| 279 | t.Fatalf("good build %q after Close, want v2: the coalesced attempt's outcome", g) |
| 280 | } |
| 281 | queue("v3") |
| 282 | if !h.m.CoalesceRecycle("bot") { |
| 283 | t.Fatal("a recycle after Close went ahead: want it skipped") |
| 284 | } |
| 285 | time.Sleep(100 * time.Millisecond) |
| 286 | if r := h.act.replacements(); len(r) != 1 { |
| 287 | t.Fatalf("replacements %+v: an attempt was admitted by a recycle after Close", r) |
| 288 | } |
| 289 | } |
| 290 | |
amadan