| 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 | // Task 5, plan review #1: the scheduler ends a planned first wake's |
| 14 | // write-ahead intent only on that wake's own terminal event, never on the |
| 15 | // activator letting the host go (a failed wake's entry goes before its |
| 16 | // outcome is delivered). So every planned wake that carries a token ends |
| 17 | // with exactly one token-carrying terminal event. The exits that sent none |
| 18 | // before now send GenCancelled: not Executed when the plan's build never |
| 19 | // ran (refused at the lease, at its check, replaced by a replan), Executed |
| 20 | // when it ran and its lifecycle ended it with no verdict. None of them is an |
| 21 | // outcome the fence must cover: they record nothing, so a new wake may plan |
| 22 | // before they arrive. |
| 23 | |
| 24 | // terminal reports the token's events that settle its build's fate (the |
| 25 | // scheduler's End): every kind but a first wake's healthy. |
| 26 | func terminal(evs []GenEvent, tok uint64) []GenEvent { |
| 27 | var out []GenEvent |
| 28 | for _, ev := range evs { |
| 29 | if ev.Token == tok && (ev.Kind != GenHealthy || !ev.FirstWake) { |
| 30 | out = append(out, ev) |
| 31 | } |
| 32 | } |
| 33 | return out |
| 34 | } |
| 35 | |
| 36 | // settle waits until the activator has gone quiet: events arrive on the |
| 37 | // wake's goroutine after the request returns. |
| 38 | func settle() { time.Sleep(100 * time.Millisecond) } |
| 39 | |
| 40 | func TestAPlannedWakeRefusedAtItsLeaseIsCancelledUnrun(t *testing.T) { |
| 41 | dir := t.TempDir() |
| 42 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 43 | a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Token: 51, ConfigRev: "c1"}) |
| 44 | var fenced atomic.Bool |
| 45 | a.OnGen = func(ev GenEvent) { |
| 46 | a.mu.Lock() |
| 47 | _, held := a.outcomeFences["one.example.com"] |
| 48 | a.mu.Unlock() |
| 49 | fenced.Store(fenced.Load() || held) |
| 50 | evs.add(ev) |
| 51 | } |
| 52 | b.mu.Lock() |
| 53 | b.acquireErr = errors.New("lease held by a previous incarnation") |
| 54 | b.mu.Unlock() |
| 55 | if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil { |
| 56 | t.Fatal("the wake was served without its lease") |
| 57 | } |
| 58 | settle() |
| 59 | got := terminal(evs.all(), 51) |
| 60 | if len(got) != 1 || got[0].Kind != GenCancelled || got[0].Executed || got[0].ConfigRev != "c1" { |
| 61 | t.Fatalf("events %+v: want one GenCancelled, not executed, with the plan's token", evs.all()) |
| 62 | } |
| 63 | if len(evs.all()) != 1 { |
| 64 | t.Fatalf("events %+v: want the one", evs.all()) |
| 65 | } |
| 66 | if fenced.Load() { |
| 67 | t.Fatal("an event that records nothing was delivered under the outcome fence") |
| 68 | } |
| 69 | } |
| 70 | |
| 71 | // A plan refused by its check before the exec, whose replan's check refuses |
| 72 | // too: both plans are cancelled unrun, each once, by its own token. |
| 73 | func TestARefusedReplanCancelsBothPlans(t *testing.T) { |
| 74 | t.Parallel() |
| 75 | dir := t.TempDir() |
| 76 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 77 | a, b, evs := planActivator(t, WakePlan{}) |
| 78 | refuse := func(string) error { return errors.New("held") } |
| 79 | var plans atomic.Uint64 |
| 80 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 81 | return WakePlan{Boot: next, Mode: ModeHold, FirstWake: true, Fallback: &good, Check: refuse, Token: 60 + plans.Add(1)} |
| 82 | } |
| 83 | if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); !errors.Is(err, errPlanStale) { |
| 84 | t.Fatalf("err = %v, want the stale plan", err) |
| 85 | } |
| 86 | settle() |
| 87 | if gens := b.started(); len(gens) != 0 { |
| 88 | t.Fatalf("spawned %+v", gens) |
| 89 | } |
| 90 | all := evs.all() |
| 91 | if len(all) != 2 { |
| 92 | t.Fatalf("events %+v: want one per plan", all) |
| 93 | } |
| 94 | for _, tok := range []uint64{61, 62} { |
| 95 | got := terminal(all, tok) |
| 96 | if len(got) != 1 || got[0].Kind != GenCancelled || got[0].Executed { |
| 97 | t.Fatalf("token %d: events %+v, want one GenCancelled, not executed", tok, all) |
| 98 | } |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // A plan replaced by a replan that boots: the replaced plan is cancelled |
| 103 | // unrun, and the replan's own events follow. |
| 104 | func TestAReplacedPlanIsCancelledUnrun(t *testing.T) { |
| 105 | t.Parallel() |
| 106 | dir := t.TempDir() |
| 107 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 108 | a, b, evs := planActivator(t, WakePlan{}) |
| 109 | held := func(build string) error { |
| 110 | if build == "v2" { |
| 111 | return errors.New("v2 failed on a sibling") |
| 112 | } |
| 113 | return nil |
| 114 | } |
| 115 | var plans atomic.Int32 |
| 116 | a.Planner = func(context.Context, registry.Route) WakePlan { |
| 117 | if plans.Add(1) == 1 { |
| 118 | return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: held, Token: 71} |
| 119 | } |
| 120 | return WakePlan{Boot: good, Mode: ModeOverlap, Check: held, Token: 72} |
| 121 | } |
| 122 | mustServe(t, a) |
| 123 | settle() |
| 124 | if gens := b.started(); len(gens) != 1 || gens[0].Build != "v1" { |
| 125 | t.Fatalf("spawned %+v, want the good v1 alone", gens) |
| 126 | } |
| 127 | old := terminal(evs.all(), 71) |
| 128 | if len(old) != 1 || old[0].Kind != GenCancelled || old[0].Executed { |
| 129 | t.Fatalf("replaced plan's events %+v: want one GenCancelled, not executed", evs.all()) |
| 130 | } |
| 131 | cur := terminal(evs.all(), 72) |
| 132 | if len(cur) != 1 || cur[0].Kind != GenHealthy { |
| 133 | t.Fatalf("replan's events %+v: want its healthy wake", evs.all()) |
| 134 | } |
| 135 | } |
| 136 | |
| 137 | // A planned wake whose worker token could not be minted reports no |
| 138 | // wokenInstance at all; it still ends with its cancellation. And the flag |
| 139 | // off (no OnGen) hears nothing, whatever the plan carries. |
| 140 | func TestAWakeWithNoResultIsStillCancelled(t *testing.T) { |
| 141 | t.Parallel() |
| 142 | plan := WakePlan{Boot: stagedID(t, t.TempDir(), "v2"), Mode: ModeHold, FirstWake: true, Token: 81} |
| 143 | a, _, evs := planActivator(t, plan) |
| 144 | e := &actEntry{host: "one.example.com", route: testRoute("one.example.com"), plan: plan} |
| 145 | a.afterWake(e, nil, errors.New("start: no token")) |
| 146 | if got := terminal(evs.all(), 81); len(got) != 1 || got[0].Kind != GenCancelled || got[0].Executed { |
| 147 | t.Fatalf("events %+v: want one GenCancelled, not executed", evs.all()) |
| 148 | } |
| 149 | a.OnGen = nil |
| 150 | a.afterWake(e, nil, errors.New("start: no token")) // must not panic or emit |
| 151 | a.afterWake(e, &wokenInstance{}, errors.New("refused")) |
| 152 | if len(evs.all()) != 1 { |
| 153 | t.Fatalf("events %+v after OnGen was unset", evs.all()) |
| 154 | } |
| 155 | } |
| 156 | |
amadan