| new file mode 100644 |
| index 000000000..a3e7c1025 |
| --- /dev/null |
| +++ b/internal/activator/swap_review_test.go |
| @@ -0,0 +1,218 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "net" |
| + "net/http" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// Index review focus 2. The agent restarts while a successor is spawning or |
| +// while the retiring generation is still draining. Shutdown joins the whole |
| +// operation set before the lease goes back. The next agent's first wake |
| +// boots the installed build through the live link, and nothing of the swap |
| +// survives the restart. This is the witness for mutations M1, M2 and M4. |
| +func TestSwapWitnessAgentRestartMidSwap(t *testing.T) { |
| + restart := func(t *testing.T, f *swapFixture) { |
| + t.Helper() |
| + f.b.execVerFn = func() string { return "v2" } // root installed v2 |
| + gensBefore, _, _ := f.b.calls() |
| + a2 := New(Config{Bucket: "test-bucket", Idle: time.Hour}, f.b, f.reg) |
| + a2.gens = true |
| + arrangeAwake(t, a2, testRoute(f.host)) |
| + if got := f.b.startCount(); got != 2 { |
| + t.Fatalf("Start (live link) calls = %d after the restart, want 2", got) |
| + } |
| + if gens, _, _ := f.b.calls(); gens != gensBefore { |
| + t.Fatal("the restarted agent spawned a generation from the old swap") |
| + } |
| + a2.mu.Lock() |
| + ver := a2.entries[f.host].spawnVersion |
| + a2.mu.Unlock() |
| + if ver != "v2" { |
| + t.Fatalf("the next wake runs %q, want the installed v2", ver) |
| + } |
| + } |
| + |
| + t.Run("successor spawning", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.b.genGate = make(chan struct{}) // never opened: only cancellation ends the spawn |
| + incumbent, rep := f.b.lastProc(), f.b.lastRep() |
| + var leaked atomic.Bool |
| + rep.setOnClose(func() { |
| + if !incumbent.Exited() { |
| + leaked.Store(true) |
| + } |
| + }) |
| + results := make(chan SwapResult, 1) |
| + go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| + waitFor(t, "the successor's spawn", func() bool { return indexPrefix(f.b.events(), "startGen:") >= 0 }) |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + f.a.Shutdown(ctx) |
| + cancel() |
| + if res := <-results; res.Outcome != SwapAbandoned { |
| + t.Fatalf("Swap = %+v, want abandoned by the shutdown", res) |
| + } |
| + if !rep.isClosed() || leaked.Load() { |
| + t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load()) |
| + } |
| + if _, raises, restores := f.b.calls(); raises != restores { |
| + t.Fatalf("raises %d restores %d", raises, restores) |
| + } |
| + restart(t, f) |
| + }) |
| + |
| + t.Run("retiring generation draining", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + incumbent, rep := f.b.lastProc(), f.b.lastRep() |
| + gate := make(chan struct{}) |
| + incumbent.setStopGate(gate) |
| + var leaked, opened atomic.Bool |
| + rep.setOnClose(func() { |
| + _, succ := f.b.lastGen() |
| + if !opened.Load() || !incumbent.Exited() || succ == nil || !succ.Exited() { |
| + leaked.Store(true) |
| + } |
| + }) |
| + results := make(chan SwapResult, 1) |
| + go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| + waitFor(t, "the switch", func() bool { return indexPrefix(f.b.events(), "switch:") >= 0 }) |
| + shut := make(chan struct{}) |
| + go func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + defer cancel() |
| + f.a.Shutdown(ctx) |
| + close(shut) |
| + }() |
| + // Shutdown has reached the entry (it is stopping) while the swap's |
| + // retire of the incumbent still sits on its gate: both are observed, |
| + // not waited out. The lease hook above is the sound half of the |
| + // negative: it flags any release made before the gate opened. |
| + waitFor(t, "Shutdown to reach the entry", func() bool { |
| + f.a.mu.Lock() |
| + defer f.a.mu.Unlock() |
| + return f.a.entries[f.host].state == actStopping |
| + }) |
| + waitFor(t, "the retire of the incumbent", func() bool { return incumbent.stopEntered() >= 1 }) |
| + // The barrier holds Shutdown's own stop of the entry back until the |
| + // retiring generation has drained. The negative is watched for a |
| + // window, since nothing signals the barrier itself: a stop that |
| + // passed it would stop the serving successor within a few |
| + // instructions, and that poll fails at once. |
| + for deadline := time.Now().Add(200 * time.Millisecond); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { |
| + if _, succ := f.b.lastGen(); succ.isStopped() { |
| + t.Fatal("Shutdown stopped the successor while the retiring generation was still draining") |
| + } |
| + if rep.isClosed() { |
| + t.Fatal("the lease went back while the retiring generation was still draining") |
| + } |
| + } |
| + select { |
| + case <-shut: |
| + t.Fatal("Shutdown returned with a generation still draining") |
| + default: |
| + } |
| + opened.Store(true) |
| + close(gate) |
| + <-shut |
| + if res := <-results; res.Outcome != SwapDone { |
| + t.Fatalf("Swap = %+v: the switch happened before the shutdown", res) |
| + } |
| + if !rep.isClosed() || leaked.Load() { |
| + t.Fatalf("lease closed %v, released while a generation ran %v", rep.isClosed(), leaked.Load()) |
| + } |
| + restart(t, f) |
| + }) |
| +} |
| + |
| +// Index review focus 4. An app that never answers GET / below 500 fails its |
| +// health check. The incumbent keeps serving and no request is dropped. The |
| +// health check is the real one, over real unix sockets, under a steady load |
| +// that dials AppSocket for every request. |
| +func TestSwapNeverHealthyAppKeepsTheIncumbentServing(t *testing.T) { |
| + dir, err := os.MkdirTemp("", "zs") // TMPDIR=/tmp/k keeps socket paths short |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + t.Cleanup(func() { os.RemoveAll(dir) }) |
| + serve := func(path string, code int) func() { |
| + ln, err := net.Listen("unix", path) |
| + if err != nil { |
| + t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/k", err) |
| + } |
| + srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(code) })} |
| + go srv.Serve(ln) //nolint:errcheck |
| + return func() { srv.Close() } |
| + } |
| + old := spawnHealthTimeout |
| + spawnHealthTimeout = 600 * time.Millisecond |
| + t.Cleanup(func() { spawnHealthTimeout = old }) |
| + |
| + f := newSwapFixture(t) |
| + f.a.genHealth = nil // the real check: GET / through the generation's handle |
| + incumbentSock := filepath.Join(dir, "inc.sock") |
| + stopIncumbent := serve(incumbentSock, http.StatusOK) |
| + defer stopIncumbent() |
| + f.incumbent.Socket, f.incumbent.Control = incumbentSock, incumbentSock+".control" |
| + setCurrentGen(t, f.a, f.host, f.incumbent) |
| + f.b.genSocketDir = dir |
| + f.b.genServe = func(g Gen) func() { return serve(g.Socket, http.StatusServiceUnavailable) } |
| + |
| + var sent, bad atomic.Int64 |
| + stop := make(chan struct{}) |
| + var wg sync.WaitGroup |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + for { |
| + select { |
| + case <-stop: |
| + return |
| + default: |
| + } |
| + sock, ok := f.a.AppSocket(f.host) |
| + sent.Add(1) |
| + if !ok { |
| + bad.Add(1) |
| + continue |
| + } |
| + resp, err := socketClient(sock).Get("http://carlos/") |
| + if err != nil { |
| + bad.Add(1) |
| + continue |
| + } |
| + if resp.StatusCode != http.StatusOK { |
| + bad.Add(1) |
| + } |
| + resp.Body.Close() |
| + } |
| + }() |
| + res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| + close(stop) |
| + wg.Wait() |
| + |
| + if res.Outcome != SwapFailed || !strings.Contains(res.Reason, "health") { |
| + t.Fatalf("Swap = %+v, want failed on health", res) |
| + } |
| + if got := f.appSocket(t); got != incumbentSock { |
| + t.Fatalf("AppSocket = %q, want the incumbent", got) |
| + } |
| + if bad.Load() != 0 || sent.Load() < 20 { |
| + t.Fatalf("%d of %d requests failed (want 0 of at least 20)", bad.Load(), sent.Load()) |
| + } |
| + if _, p := f.b.lastGen(); p == nil || !p.isStopped() { |
| + t.Fatal("the never-healthy successor was left running") |
| + } |
| + if f.b.lastProc().isStopped() { |
| + t.Fatal("the incumbent was stopped") |
| + } |
| + if _, raises, restores := f.b.calls(); raises != 1 || restores != 1 { |
| + t.Fatalf("raises %d restores %d, want 1 and 1", raises, restores) |
| + } |
| +} |
amadan