| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "sync/atomic" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | ) |
| 12 | |
| 13 | // Final review F4 (activator half): every GenEvent of a planned wake carries |
| 14 | // that wake's token and the config revision it started with, as the plan |
| 15 | // handed them in (WakePlan.Token, WakePlan.ConfigRev), so the scheduler can |
| 16 | // match an outcome to the wake that produced it and an old event can never |
| 17 | // clear or activate a newer wake's state. Covered: afterWake's events |
| 18 | // (healthy, fellback, failed, untried, a worker's settled) and the |
| 19 | // observer's (settled, failed, cancelled), and a replanned wake, whose |
| 20 | // events carry the replan's token. |
| 21 | func TestEveryGenEventCarriesItsWakesToken(t *testing.T) { |
| 22 | const token, rev = uint64(0x5eed), "cfg-7" |
| 23 | for _, tc := range []struct { |
| 24 | name string |
| 25 | run func(t *testing.T) []GenEvent |
| 26 | }{ |
| 27 | {"healthy, then settled by the observer", func(t *testing.T) []GenEvent { |
| 28 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 29 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 30 | a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}) |
| 31 | mustServe(t, a) |
| 32 | evs.wait(t, GenSettled) |
| 33 | return evs.all() |
| 34 | }}, |
| 35 | {"fell back", func(t *testing.T) []GenEvent { |
| 36 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 37 | a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}) |
| 38 | a.genHealth = failBuild("v2") |
| 39 | mustServe(t, a) |
| 40 | evs.wait(t, GenFellBack) |
| 41 | return evs.all() |
| 42 | }}, |
| 43 | {"failed in hold mode", func(t *testing.T) []GenEvent { |
| 44 | next := stagedID(t, t.TempDir(), "v2") |
| 45 | a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true, Token: token, ConfigRev: rev}) |
| 46 | a.genHealth = failBuild("v2") |
| 47 | _, _ = a.beginRequest(context.Background(), testRoute("one.example.com")) |
| 48 | evs.wait(t, GenFailed) |
| 49 | return evs.all() |
| 50 | }}, |
| 51 | {"untried under the verdict floor", func(t *testing.T) []GenEvent { |
| 52 | shrinkVerdictFloor(t, time.Hour) |
| 53 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 54 | a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}) |
| 55 | mustServe(t, a) |
| 56 | evs.wait(t, GenHealthy) |
| 57 | return evs.all() |
| 58 | }}, |
| 59 | {"failed by the observer", func(t *testing.T) []GenEvent { |
| 60 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 61 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 62 | a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}) |
| 63 | mustServe(t, a) |
| 64 | evs.wait(t, GenHealthy) |
| 65 | b.procOf(b.started()[0].N).crash() |
| 66 | evs.wait(t, GenFailed) |
| 67 | return evs.all() |
| 68 | }}, |
| 69 | {"cancelled by an intentional stop", func(t *testing.T) []GenEvent { |
| 70 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 71 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 72 | a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev}) |
| 73 | mustServe(t, a) |
| 74 | evs.wait(t, GenHealthy) |
| 75 | a.BounceFor("one.example.com", CauseConfig) |
| 76 | evs.wait(t, GenCancelled) |
| 77 | return evs.all() |
| 78 | }}, |
| 79 | {"a worker's first wake, settled at its readiness", func(t *testing.T) []GenEvent { |
| 80 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| 81 | a, _, _, _, _ := stagedActivator(t) |
| 82 | a.gens = true |
| 83 | evs := &genLog{} |
| 84 | a.OnGen = evs.add |
| 85 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 86 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 87 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: token, ConfigRev: rev} |
| 88 | } |
| 89 | if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | evs.wait(t, GenSettled) |
| 93 | return evs.all() |
| 94 | }}, |
| 95 | {"replanned at the exec: the replan's token", func(t *testing.T) []GenEvent { |
| 96 | good, next := stagedID(t, t.TempDir(), "v1"), stagedID(t, t.TempDir(), "v2") |
| 97 | a, _, evs := planActivator(t, WakePlan{}) |
| 98 | var plans atomic.Int32 |
| 99 | held := func(build string) error { |
| 100 | if build == "v2" { |
| 101 | return errors.New("v2 failed on a sibling") |
| 102 | } |
| 103 | return nil |
| 104 | } |
| 105 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 106 | if plans.Add(1) == 1 { |
| 107 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: held, Token: token - 1, ConfigRev: "cfg-6"} |
| 108 | } |
| 109 | return WakePlan{Boot: good, Mode: ModeOverlap, Check: held, Token: token, ConfigRev: rev} |
| 110 | } |
| 111 | mustServe(t, a) |
| 112 | evs.wait(t, GenHealthy) |
| 113 | // The replaced plan's own cancellation (Task 5) carries its own |
| 114 | // token; every other event is the replan's. |
| 115 | var out []GenEvent |
| 116 | for _, ev := range evs.all() { |
| 117 | if ev.Token == token-1 { |
| 118 | if ev.Kind != GenCancelled || ev.Executed || ev.ConfigRev != "cfg-6" { |
| 119 | t.Fatalf("replaced plan's event %+v: want GenCancelled, not executed, with its own token and config", ev) |
| 120 | } |
| 121 | continue |
| 122 | } |
| 123 | out = append(out, ev) |
| 124 | } |
| 125 | if len(out) == len(evs.all()) { |
| 126 | t.Fatalf("events %+v: the replaced plan was not cancelled", evs.all()) |
| 127 | } |
| 128 | return out |
| 129 | }}, |
| 130 | } { |
| 131 | t.Run(tc.name, func(t *testing.T) { |
| 132 | evs := tc.run(t) |
| 133 | if len(evs) == 0 { |
| 134 | t.Fatal("no events") |
| 135 | } |
| 136 | for _, ev := range evs { |
| 137 | if ev.Token != token || ev.ConfigRev != rev { |
| 138 | t.Fatalf("%s event carries token %#x, config %q; want the wake's %#x, %q (all: %+v)", ev.Kind, ev.Token, ev.ConfigRev, token, rev, evs) |
| 139 | } |
| 140 | } |
| 141 | }) |
| 142 | } |
| 143 | } |
| 144 | |
| 145 | func mustServe(t *testing.T, a *Service) { |
| 146 | t.Helper() |
| 147 | done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| 148 | if err != nil { |
| 149 | t.Fatal(err) |
| 150 | } |
| 151 | done() |
| 152 | } |
| 153 | |
| 154 | func failBuild(build string) func(context.Context, Gen) error { |
| 155 | return func(_ context.Context, g Gen) error { |
| 156 | if g.Build == build { |
| 157 | return errors.New("GET / answered 500") |
| 158 | } |
| 159 | return nil |
| 160 | } |
| 161 | } |
| 162 | |
| 163 | func (l *genLog) all() []GenEvent { |
| 164 | l.mu.Lock() |
| 165 | defer l.mu.Unlock() |
| 166 | return append([]GenEvent(nil), l.evs...) |
| 167 | } |
| 168 | |
amadan