| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "slices" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "sync/atomic" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/registry" |
| 14 | ) |
| 15 | |
| 16 | // blockedOnGen is an OnGen that blocks on events of kind until release |
| 17 | // closes, as a scheduler whose failure transition is slow would, and marks |
| 18 | // recorded once that event's handling has returned. Only the first such |
| 19 | // event is the one a test waits on. |
| 20 | type blockedOnGen struct { |
| 21 | kind GenEventKind |
| 22 | once sync.Once |
| 23 | entered chan struct{} |
| 24 | release chan struct{} |
| 25 | recorded atomic.Bool |
| 26 | evs genLog |
| 27 | } |
| 28 | |
| 29 | func newBlockedOnGen(kind GenEventKind) *blockedOnGen { |
| 30 | return &blockedOnGen{kind: kind, entered: make(chan struct{}), release: make(chan struct{})} |
| 31 | } |
| 32 | |
| 33 | func (o *blockedOnGen) on(ev GenEvent) { |
| 34 | o.evs.add(ev) |
| 35 | if ev.Kind != o.kind { |
| 36 | return |
| 37 | } |
| 38 | o.once.Do(func() { close(o.entered) }) |
| 39 | <-o.release |
| 40 | o.recorded.Store(true) |
| 41 | } |
| 42 | |
| 43 | func (o *blockedOnGen) waitEntered(t *testing.T) { |
| 44 | t.Helper() |
| 45 | select { |
| 46 | case <-o.entered: |
| 47 | case <-time.After(5 * time.Second): |
| 48 | t.Fatalf("OnGen never received %s; saw %v", o.kind, o.evs.kinds()) |
| 49 | } |
| 50 | } |
| 51 | |
| 52 | // secondPlanner answers the first plan with first, and records, for every |
| 53 | // later plan, whether the blocked outcome had been recorded when it was |
| 54 | // asked. A later plan refuses once it has, as the scheduler's planner does |
| 55 | // for a held-down build. |
| 56 | func secondPlanner(o *blockedOnGen, first WakePlan, saw chan<- bool) func(context.Context, registry.Route) WakePlan { |
| 57 | var n atomic.Int32 |
| 58 | return func(context.Context, registry.Route) WakePlan { |
| 59 | if n.Add(1) == 1 { |
| 60 | return first |
| 61 | } |
| 62 | rec := o.recorded.Load() |
| 63 | saw <- rec |
| 64 | if rec { |
| 65 | return WakePlan{Refuse: "held down"} |
| 66 | } |
| 67 | return first |
| 68 | } |
| 69 | } |
| 70 | |
| 71 | // Final review F3 (spec §8.1, O7): a failed first wake's outcome is |
| 72 | // delivered before admission reopens. A request that arrives after the |
| 73 | // failed wake's entry is gone, while OnGen is still recording the failure, |
| 74 | // starts a wake that waits for that delivery and only then plans, with the |
| 75 | // record in place. Before the fence the second wake planned B again in the |
| 76 | // gap, before the hold-down existed. |
| 77 | func TestANewWakeWaitsForAFailedFirstWakesOutcome(t *testing.T) { |
| 78 | t.Parallel() |
| 79 | dir := t.TempDir() |
| 80 | next := stagedID(t, dir, "v2") |
| 81 | a, b, _ := planActivator(t, WakePlan{}) |
| 82 | const host = "one.example.com" |
| 83 | a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") } |
| 84 | o := newBlockedOnGen(GenFailed) |
| 85 | a.OnGen = o.on |
| 86 | saw := make(chan bool, 4) |
| 87 | a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw) |
| 88 | if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| 89 | t.Fatal("the failed hold first wake served") |
| 90 | } |
| 91 | o.waitEntered(t) |
| 92 | if a.Owns(host) { |
| 93 | t.Fatal("fixture: the failed wake's entry is still there") |
| 94 | } |
| 95 | done := make(chan error, 1) |
| 96 | go func() { |
| 97 | _, err := a.beginRequest(context.Background(), testRoute(host)) |
| 98 | done <- err |
| 99 | }() |
| 100 | select { |
| 101 | case rec := <-saw: |
| 102 | close(o.release) |
| 103 | t.Fatalf("the second wake planned while the first wake's failure was still being delivered (record in place: %v)", rec) |
| 104 | case <-time.After(300 * time.Millisecond): |
| 105 | } |
| 106 | close(o.release) |
| 107 | select { |
| 108 | case rec := <-saw: |
| 109 | if !rec { |
| 110 | t.Fatal("the second wake planned before the failure was recorded") |
| 111 | } |
| 112 | case <-time.After(5 * time.Second): |
| 113 | t.Fatal("the second wake never planned") |
| 114 | } |
| 115 | if err := <-done; err == nil || !strings.Contains(err.Error(), "held down") { |
| 116 | t.Fatalf("second wake = %v, want refused by the recorded hold-down", err) |
| 117 | } |
| 118 | if gens := b.started(); len(gens) != 1 { |
| 119 | t.Fatalf("spawned %+v: B ran again before its failure was recorded", gens) |
| 120 | } |
| 121 | } |
| 122 | |
| 123 | // Final review wave 2 (N1): the fence fails closed. The wait is bounded |
| 124 | // (outcomeFenceWait), the restriction is not. A wake still behind an |
| 125 | // undelivered outcome when the bound passes is refused before it plans: |
| 126 | // no Planner call, nothing launched, and the member reads the maintenance |
| 127 | // fence's existing line. The fence stays, so the next request is refused |
| 128 | // the same way. Once the outcome is delivered a wake proceeds to plan |
| 129 | // (here the planner, with the record in place, refuses the held build). |
| 130 | // Before the fix the wake planned after the bound and could boot B again |
| 131 | // without its record. |
| 132 | func TestTheOutcomeFenceFailsClosed(t *testing.T) { |
| 133 | old := outcomeFenceWait |
| 134 | outcomeFenceWait = 200 * time.Millisecond |
| 135 | t.Cleanup(func() { outcomeFenceWait = old }) |
| 136 | dir := t.TempDir() |
| 137 | next := stagedID(t, dir, "v2") |
| 138 | a, b, _ := planActivator(t, WakePlan{}) |
| 139 | const host = "one.example.com" |
| 140 | a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") } |
| 141 | o := newBlockedOnGen(GenFailed) |
| 142 | var releaseOnce sync.Once |
| 143 | release := func() { releaseOnce.Do(func() { close(o.release) }) } |
| 144 | t.Cleanup(release) |
| 145 | a.OnGen = o.on |
| 146 | sink := &lockedSink{} |
| 147 | a.Logs = sink |
| 148 | saw := make(chan bool, 4) |
| 149 | a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true}, saw) |
| 150 | if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| 151 | t.Fatal("the failed hold first wake served") |
| 152 | } |
| 153 | o.waitEntered(t) |
| 154 | for i := 1; i <= 2; i++ { |
| 155 | start := time.Now() |
| 156 | _, err := a.beginRequest(context.Background(), testRoute(host)) |
| 157 | if !errors.Is(err, errOutcomeUndelivered) { |
| 158 | t.Fatalf("wake %d behind the undelivered outcome = %v, want refused (errOutcomeUndelivered)", i, err) |
| 159 | } |
| 160 | if took := time.Since(start); took < 150*time.Millisecond { |
| 161 | t.Fatalf("wake %d refused after %v: it did not wait for the fence", i, took) |
| 162 | } |
| 163 | select { |
| 164 | case rec := <-saw: |
| 165 | t.Fatalf("wake %d planned behind the undelivered outcome (record in place: %v)", i, rec) |
| 166 | default: |
| 167 | } |
| 168 | if gens := b.started(); len(gens) != 1 { |
| 169 | t.Fatalf("spawned %+v: a wake behind the undelivered outcome launched something", gens) |
| 170 | } |
| 171 | } |
| 172 | want := memberSafeWakeReason(ErrFenced) |
| 173 | got := sink.errors() |
| 174 | if len(got) != 3 || got[1] != want || got[2] != want { |
| 175 | t.Fatalf("member error events %q: want the failed wake's, then the maintenance fence's line (%q) for each refusal", got, want) |
| 176 | } |
| 177 | release() |
| 178 | _, err := a.beginRequest(context.Background(), testRoute(host)) |
| 179 | select { |
| 180 | case rec := <-saw: |
| 181 | if !rec { |
| 182 | t.Fatal("the wake after the delivery planned before the failure was recorded") |
| 183 | } |
| 184 | case <-time.After(5 * time.Second): |
| 185 | t.Fatal("the wake after the delivery never planned: the fence outlived the outcome") |
| 186 | } |
| 187 | if err == nil || !strings.Contains(err.Error(), "held down") { |
| 188 | t.Fatalf("wake after the delivery = %v, want refused by the recorded hold-down", err) |
| 189 | } |
| 190 | if gens := b.started(); len(gens) != 1 { |
| 191 | t.Fatalf("spawned %+v: B ran again", gens) |
| 192 | } |
| 193 | } |
| 194 | |
| 195 | // Final review F3, the observer's half: a first wake's 10 s observation |
| 196 | // delivers its terminal event (here GenFailed, an exit inside the window) |
| 197 | // BEFORE it clears firstWatch, and holds the host's outcome fence until |
| 198 | // the delivery returns. While OnGen records the failure, Current still |
| 199 | // hides the generation, and a request that stops the dead entry and wakes |
| 200 | // it again waits for the record before planning. Before the fix the |
| 201 | // observer cleared firstWatch first and nothing held the next wake back. |
| 202 | func TestAFirstWakeObserverDeliversItsOutcomeBeforeAdmissionReopens(t *testing.T) { |
| 203 | shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| 204 | dir := t.TempDir() |
| 205 | good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| 206 | a, b, _ := planActivator(t, WakePlan{}) |
| 207 | const host = "one.example.com" |
| 208 | o := newBlockedOnGen(GenFailed) |
| 209 | a.OnGen = o.on |
| 210 | saw := make(chan bool, 4) |
| 211 | a.Planner = secondPlanner(o, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}, saw) |
| 212 | done, err := a.beginRequest(context.Background(), testRoute(host)) |
| 213 | if err != nil { |
| 214 | t.Fatal(err) |
| 215 | } |
| 216 | done() |
| 217 | o.evs.wait(t, GenHealthy) |
| 218 | b.procOf(b.started()[0].N).crash() |
| 219 | o.waitEntered(t) |
| 220 | if _, ok := a.Current(host); ok { |
| 221 | close(o.release) |
| 222 | t.Fatal("Current showed the first wake while its failure was still being delivered: firstWatch was cleared first") |
| 223 | } |
| 224 | second := make(chan error, 1) |
| 225 | go func() { |
| 226 | _, err := a.beginRequest(context.Background(), testRoute(host)) |
| 227 | second <- err |
| 228 | }() |
| 229 | select { |
| 230 | case rec := <-saw: |
| 231 | close(o.release) |
| 232 | t.Fatalf("the next wake planned while the observer's failure was still being delivered (record in place: %v)", rec) |
| 233 | case <-time.After(300 * time.Millisecond): |
| 234 | } |
| 235 | close(o.release) |
| 236 | select { |
| 237 | case rec := <-saw: |
| 238 | if !rec { |
| 239 | t.Fatal("the next wake planned before the failure was recorded") |
| 240 | } |
| 241 | case <-time.After(5 * time.Second): |
| 242 | t.Fatal("the next wake never planned") |
| 243 | } |
| 244 | if err := <-second; err == nil || !strings.Contains(err.Error(), "held down") { |
| 245 | t.Fatalf("next wake = %v, want refused by the recorded failure", err) |
| 246 | } |
| 247 | if k := o.evs.kinds(); !slices.Equal(k, []GenEventKind{GenHealthy, GenFailed}) { |
| 248 | t.Fatalf("events %v, want healthy then failed", k) |
| 249 | } |
| 250 | } |
| 251 | |
amadan