| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "net" |
| 6 | "net/http" |
| 7 | "os" |
| 8 | "path/filepath" |
| 9 | "strings" |
| 10 | "sync" |
| 11 | "sync/atomic" |
| 12 | "testing" |
| 13 | "time" |
| 14 | ) |
| 15 | |
| 16 | // Index review focus 2. The agent restarts while a successor is spawning or |
| 17 | // while the retiring generation is still draining. Shutdown joins the whole |
| 18 | // operation set before the lease goes back. The next agent's first wake |
| 19 | // boots the installed build through the live link, and nothing of the swap |
| 20 | // survives the restart. This is the witness for mutations M1, M2 and M4. |
| 21 | func TestSwapWitnessAgentRestartMidSwap(t *testing.T) { |
| 22 | t.Parallel() |
| 23 | restart := func(t *testing.T, f *swapFixture) { |
| 24 | t.Helper() |
| 25 | f.b.execVerFn = func() string { return "v2" } // root installed v2 |
| 26 | gensBefore, _, _ := f.b.calls() |
| 27 | a2 := New(Config{Bucket: "test-bucket", Idle: time.Hour}, f.b, f.reg) |
| 28 | a2.gens = true |
| 29 | arrangeAwake(t, a2, testRoute(f.host)) |
| 30 | if got := f.b.startCount(); got != 2 { |
| 31 | t.Fatalf("Start (live link) calls = %d after the restart, want 2", got) |
| 32 | } |
| 33 | if gens, _, _ := f.b.calls(); gens != gensBefore { |
| 34 | t.Fatal("the restarted agent spawned a generation from the old swap") |
| 35 | } |
| 36 | a2.mu.Lock() |
| 37 | ver := a2.entries[f.host].spawnVersion |
| 38 | a2.mu.Unlock() |
| 39 | if ver != "v2" { |
| 40 | t.Fatalf("the next wake runs %q, want the installed v2", ver) |
| 41 | } |
| 42 | } |
| 43 | |
| 44 | t.Run("successor spawning", func(t *testing.T) { |
| 45 | f := newSwapFixture(t) |
| 46 | // A spawn that outlives cancellation: StartStaged ignores its ctx |
| 47 | // until the hold opens, then returns a live successor. Shutdown |
| 48 | // cancels the attempt, but only the join makes it wait for that |
| 49 | // process, so the incumbent may not stop and the lease may not go |
| 50 | // back until the successor exists and has been stopped. |
| 51 | hold := make(chan struct{}) |
| 52 | f.b.genHold = hold |
| 53 | incumbent, rep := f.b.lastProc(), f.b.lastRep() |
| 54 | var leaked atomic.Bool |
| 55 | rep.setOnClose(func() { |
| 56 | if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() || !incumbent.Exited() { |
| 57 | leaked.Store(true) |
| 58 | } |
| 59 | }) |
| 60 | results := make(chan SwapResult, 1) |
| 61 | go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 62 | waitFor(t, "the successor's spawn", func() bool { return indexPrefix(f.b.events(), "startGen:") >= 0 }) |
| 63 | shut := make(chan struct{}) |
| 64 | go func() { |
| 65 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 66 | defer cancel() |
| 67 | f.a.Shutdown(ctx) |
| 68 | close(shut) |
| 69 | }() |
| 70 | waitFor(t, "Shutdown to reach the entry", func() bool { |
| 71 | f.a.mu.Lock() |
| 72 | defer f.a.mu.Unlock() |
| 73 | // A missing entry means the stop already finished: the checks |
| 74 | // below report that, rather than a nil dereference here. |
| 75 | e := f.a.entries[f.host] |
| 76 | return e == nil || e.state == actStopping |
| 77 | }) |
| 78 | // The barrier holds the stop back while the spawn is in flight. As in |
| 79 | // the draining case nothing signals the barrier itself, so the |
| 80 | // negative is watched for a window; a stop that passed it would stop |
| 81 | // the incumbent at once, and the poll fails then. |
| 82 | for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { |
| 83 | if incumbent.isStopped() { |
| 84 | t.Fatal("the incumbent was stopped while a successor was still spawning") |
| 85 | } |
| 86 | if rep.isClosed() { |
| 87 | t.Fatal("the lease went back while a successor was still spawning") |
| 88 | } |
| 89 | } |
| 90 | select { |
| 91 | case <-shut: |
| 92 | t.Fatal("Shutdown returned with a successor still spawning") |
| 93 | default: |
| 94 | } |
| 95 | close(hold) |
| 96 | <-shut |
| 97 | if res := <-results; res.Outcome != SwapAbandoned { |
| 98 | t.Fatalf("Swap = %+v, want abandoned by the shutdown", res) |
| 99 | } |
| 100 | if _, succ := f.b.lastGen(); succ == nil || !succ.Exited() { |
| 101 | t.Fatal("the successor that outlived the cancellation was never stopped") |
| 102 | } |
| 103 | if !rep.isClosed() || leaked.Load() { |
| 104 | t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load()) |
| 105 | } |
| 106 | if _, raises, restores := f.b.calls(); raises != restores { |
| 107 | t.Fatalf("raises %d restores %d", raises, restores) |
| 108 | } |
| 109 | restart(t, f) |
| 110 | }) |
| 111 | |
| 112 | t.Run("retiring generation draining", func(t *testing.T) { |
| 113 | f := newSwapFixture(t) |
| 114 | incumbent, rep := f.b.lastProc(), f.b.lastRep() |
| 115 | gate := make(chan struct{}) |
| 116 | incumbent.setStopGate(gate) |
| 117 | var leaked, opened atomic.Bool |
| 118 | rep.setOnClose(func() { |
| 119 | _, succ := f.b.lastGen() |
| 120 | if !opened.Load() || !incumbent.Exited() || succ == nil || !succ.Exited() { |
| 121 | leaked.Store(true) |
| 122 | } |
| 123 | }) |
| 124 | results := make(chan SwapResult, 1) |
| 125 | go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| 126 | waitFor(t, "the switch", func() bool { return indexPrefix(f.b.events(), "switch:") >= 0 }) |
| 127 | shut := make(chan struct{}) |
| 128 | go func() { |
| 129 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 130 | defer cancel() |
| 131 | f.a.Shutdown(ctx) |
| 132 | close(shut) |
| 133 | }() |
| 134 | // Shutdown has reached the entry (it is stopping) while the swap's |
| 135 | // retire of the incumbent still sits on its gate: both are observed, |
| 136 | // not waited out. The lease hook above is the sound half of the |
| 137 | // negative: it flags any release made before the gate opened. |
| 138 | waitFor(t, "Shutdown to reach the entry", func() bool { |
| 139 | f.a.mu.Lock() |
| 140 | defer f.a.mu.Unlock() |
| 141 | // A missing entry means the stop already finished: the checks |
| 142 | // below report that, rather than a nil dereference here. |
| 143 | e := f.a.entries[f.host] |
| 144 | return e == nil || e.state == actStopping |
| 145 | }) |
| 146 | waitFor(t, "the retire of the incumbent", func() bool { return incumbent.stopEntered() >= 1 }) |
| 147 | // The barrier holds Shutdown's own stop of the entry back until the |
| 148 | // retiring generation has drained. The negative is watched for a |
| 149 | // window, since nothing signals the barrier itself: a stop that |
| 150 | // passed it would stop the serving successor within a few |
| 151 | // instructions, and that poll fails at once. |
| 152 | for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { |
| 153 | if _, succ := f.b.lastGen(); succ.isStopped() { |
| 154 | t.Fatal("Shutdown stopped the successor while the retiring generation was still draining") |
| 155 | } |
| 156 | if rep.isClosed() { |
| 157 | t.Fatal("the lease went back while the retiring generation was still draining") |
| 158 | } |
| 159 | } |
| 160 | select { |
| 161 | case <-shut: |
| 162 | t.Fatal("Shutdown returned with a generation still draining") |
| 163 | default: |
| 164 | } |
| 165 | opened.Store(true) |
| 166 | close(gate) |
| 167 | <-shut |
| 168 | if res := <-results; res.Outcome != SwapDone { |
| 169 | t.Fatalf("Swap = %+v: the switch happened before the shutdown", res) |
| 170 | } |
| 171 | if !rep.isClosed() || leaked.Load() { |
| 172 | t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load()) |
| 173 | } |
| 174 | restart(t, f) |
| 175 | }) |
| 176 | } |
| 177 | |
| 178 | // Index review focus 4. An app that never answers GET / below 500 fails its |
| 179 | // health check. The incumbent keeps serving and no request is dropped. The |
| 180 | // health check is the real one, over real unix sockets, under a steady load |
| 181 | // that dials AppSocket for every request. |
| 182 | func TestSwapNeverHealthyAppKeepsTheIncumbentServing(t *testing.T) { |
| 183 | dir, err := os.MkdirTemp("", "zs") // TMPDIR=/tmp/k keeps socket paths short |
| 184 | if err != nil { |
| 185 | t.Fatal(err) |
| 186 | } |
| 187 | t.Cleanup(func() { os.RemoveAll(dir) }) |
| 188 | serve := func(path string, code int) func() { |
| 189 | ln, err := net.Listen("unix", path) |
| 190 | if err != nil { |
| 191 | t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/k", err) |
| 192 | } |
| 193 | srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(code) })} |
| 194 | go srv.Serve(ln) //nolint:errcheck |
| 195 | return func() { srv.Close() } |
| 196 | } |
| 197 | old := spawnHealthTimeout |
| 198 | spawnHealthTimeout = 600 * time.Millisecond |
| 199 | t.Cleanup(func() { spawnHealthTimeout = old }) |
| 200 | |
| 201 | f := newSwapFixture(t) |
| 202 | f.a.genHealth = nil // the real check: GET / through the generation's handle |
| 203 | incumbentSock := filepath.Join(dir, "inc.sock") |
| 204 | stopIncumbent := serve(incumbentSock, http.StatusOK) |
| 205 | defer stopIncumbent() |
| 206 | f.incumbent.Socket, f.incumbent.Control = incumbentSock, incumbentSock+".control" |
| 207 | setCurrentGen(t, f.a, f.host, f.incumbent) |
| 208 | f.b.genSocketDir = dir |
| 209 | f.b.genServe = func(g Gen) func() { return serve(g.Socket, http.StatusServiceUnavailable) } |
| 210 | |
| 211 | var sent, bad atomic.Int64 |
| 212 | stop := make(chan struct{}) |
| 213 | var wg sync.WaitGroup |
| 214 | wg.Add(1) |
| 215 | go func() { |
| 216 | defer wg.Done() |
| 217 | for { |
| 218 | select { |
| 219 | case <-stop: |
| 220 | return |
| 221 | default: |
| 222 | } |
| 223 | sock, ok := f.a.AppSocket(f.host) |
| 224 | sent.Add(1) |
| 225 | if !ok { |
| 226 | bad.Add(1) |
| 227 | continue |
| 228 | } |
| 229 | resp, err := socketClient(sock).Get("http://carlos/") |
| 230 | if err != nil { |
| 231 | bad.Add(1) |
| 232 | continue |
| 233 | } |
| 234 | if resp.StatusCode != http.StatusOK { |
| 235 | bad.Add(1) |
| 236 | } |
| 237 | resp.Body.Close() |
| 238 | } |
| 239 | }() |
| 240 | res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| 241 | close(stop) |
| 242 | wg.Wait() |
| 243 | |
| 244 | if res.Outcome != SwapFailed || !strings.Contains(res.Reason, "health") { |
| 245 | t.Fatalf("Swap = %+v, want failed on health", res) |
| 246 | } |
| 247 | if got := f.appSocket(t); got != incumbentSock { |
| 248 | t.Fatalf("AppSocket = %q, want the incumbent", got) |
| 249 | } |
| 250 | if bad.Load() != 0 || sent.Load() < 20 { |
| 251 | t.Fatalf("%d of %d requests failed (want 0 of at least 20)", bad.Load(), sent.Load()) |
| 252 | } |
| 253 | if _, p := f.b.lastGen(); p == nil || !p.isStopped() { |
| 254 | t.Fatal("the never-healthy successor was left running") |
| 255 | } |
| 256 | if f.b.lastProc().isStopped() { |
| 257 | t.Fatal("the incumbent was stopped") |
| 258 | } |
| 259 | if _, raises, restores := f.b.calls(); raises != 1 || restores != 1 { |
| 260 | t.Fatalf("raises %d restores %d, want 1 and 1", raises, restores) |
| 261 | } |
| 262 | } |
| 263 | |
amadan