| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "errors" |
| 5 | "slices" |
| 6 | "sync" |
| 7 | "testing" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // Final review F1: a worker respawn asks its ownership again at the exec |
| 15 | // (composed into the before-exec hook, after the build's publication, or |
| 16 | // right before StartWorker on a backend without the staged capability), |
| 17 | // and once more before it installs the child. A lease lost while the |
| 18 | // respawn's preparation is under way (here, held open in the config |
| 19 | // admission) execs nothing; one lost after the exec installs nothing, and |
| 20 | // the child it made is shed. With generations on or off, on a backend with |
| 21 | // or without the staged capability. |
| 22 | func TestWorkerRespawnRechecksOwnershipAtTheExecAndTheInstall(t *testing.T) { |
| 23 | t.Parallel() |
| 24 | type rig struct { |
| 25 | a *Service |
| 26 | b *fakeBackend |
| 27 | staged bool |
| 28 | } |
| 29 | setup := func(t *testing.T, on, staged bool) rig { |
| 30 | a, b, _, _ := testActivator(t, time.Hour) |
| 31 | a.gens = on |
| 32 | if staged { |
| 33 | a.backend = &stagedFake{fakeBackend: b} |
| 34 | } |
| 35 | return rig{a, b, staged} |
| 36 | } |
| 37 | for _, on := range []bool{true, false} { |
| 38 | for _, staged := range []bool{true, false} { |
| 39 | name := map[bool]string{true: "gens on", false: "gens off"}[on] + map[bool]string{true: ", staged", false: ", plain"}[staged] |
| 40 | t.Run(name+": lost during the preparation", func(t *testing.T) { |
| 41 | r := setup(t, on, staged) |
| 42 | rt := testWorkerRoute("bot6.example.com") |
| 43 | if err := r.a.WakeWorker(rt); err != nil { |
| 44 | t.Fatal(err) |
| 45 | } |
| 46 | old := r.b.lastWorker() |
| 47 | rep := r.b.lastRep() |
| 48 | entered, release := make(chan struct{}), make(chan struct{}) |
| 49 | var once sync.Once |
| 50 | r.b.mu.Lock() |
| 51 | r.b.admitHook = func() { |
| 52 | once.Do(func() { close(entered) }) |
| 53 | <-release |
| 54 | } |
| 55 | r.b.mu.Unlock() |
| 56 | r.a.mu.Lock() |
| 57 | r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails mid-preparation", nil) |
| 58 | r.a.mu.Unlock() |
| 59 | <-entered // past the first ownership check, inside the preparation |
| 60 | rep.setErr(errors.New("renewal failed")) |
| 61 | close(release) |
| 62 | waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) }) |
| 63 | if got := r.b.workerEnters(); got != 1 { |
| 64 | t.Fatalf("worker spawns begun = %d: the respawn exec'd after its lease was lost", got) |
| 65 | } |
| 66 | r.a.mu.Lock() |
| 67 | e := r.a.entries[rt.Host] |
| 68 | proc := e.proc |
| 69 | r.a.mu.Unlock() |
| 70 | if proc != Proc(old) { |
| 71 | t.Fatal("the respawn installed a child after its lease was lost") |
| 72 | } |
| 73 | if _, members := r.a.genStateForTest(rt.Host); len(members) != 0 { |
| 74 | t.Fatalf("members after the refused respawn = %v", members) |
| 75 | } |
| 76 | }) |
| 77 | t.Run(name+": lost after the exec", func(t *testing.T) { |
| 78 | r := setup(t, on, staged) |
| 79 | rt := testWorkerRoute("bot7.example.com") |
| 80 | if err := r.a.WakeWorker(rt); err != nil { |
| 81 | t.Fatal(err) |
| 82 | } |
| 83 | old := r.b.lastWorker() |
| 84 | rep := r.b.lastRep() |
| 85 | r.b.mu.Lock() |
| 86 | r.b.workerStartHook = func(_ registry.Route) { rep.setErr(errors.New("renewal failed")) } |
| 87 | r.b.mu.Unlock() |
| 88 | r.a.mu.Lock() |
| 89 | r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails at the exec", nil) |
| 90 | r.a.mu.Unlock() |
| 91 | waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) }) |
| 92 | if got := r.b.workerStarts(); got != 2 { |
| 93 | t.Fatalf("worker starts = %d, want 2 (the wake and the respawn's exec)", got) |
| 94 | } |
| 95 | fresh := r.b.lastWorker() |
| 96 | r.a.mu.Lock() |
| 97 | proc := r.a.entries[rt.Host].proc |
| 98 | r.a.mu.Unlock() |
| 99 | if proc != Proc(old) || proc == Proc(fresh) { |
| 100 | t.Fatal("the respawn installed its child after its lease was lost") |
| 101 | } |
| 102 | if !fresh.isStopped() { |
| 103 | t.Fatal("the uninstalled child was left running") |
| 104 | } |
| 105 | }) |
| 106 | } |
| 107 | } |
| 108 | } |
| 109 | |
| 110 | // entryPendingStop reports whether host's entry is a stop held pending with |
| 111 | // its lease kept. |
| 112 | func (a *Service) entryPendingStop(host string) bool { |
| 113 | a.mu.Lock() |
| 114 | defer a.mu.Unlock() |
| 115 | e := a.entries[host] |
| 116 | return e != nil && e.pendingStop |
| 117 | } |
| 118 | |
| 119 | func (a *Service) entryPresent(host string) bool { |
| 120 | a.mu.Lock() |
| 121 | defer a.mu.Unlock() |
| 122 | return a.entries[host] != nil |
| 123 | } |
| 124 | |
| 125 | // Final review R2-1: with generations off, a respawn child that exec'd and |
| 126 | // was then refused at its installation (here: the lease failed after the |
| 127 | // exec) stays the entry's to clean up until its exit is proven. Its build |
| 128 | // stays pinned, and the terminal stop that the lost lease begins keeps the |
| 129 | // lease and the entry until the child has exited, whether the shed Stop |
| 130 | // failed to prove it or is still blocked when the stop's bounded wait on |
| 131 | // the respawn runs out. |
| 132 | func TestAShedRespawnChildIsRetainedUntilItsExitIsProven(t *testing.T) { |
| 133 | for _, blocked := range []bool{false, true} { |
| 134 | name := map[bool]string{false: "the shed Stop fails", true: "the shed Stop blocks past the stop's respawn wait"}[blocked] |
| 135 | t.Run(name, func(t *testing.T) { |
| 136 | if blocked { |
| 137 | old := respawnJoinBound |
| 138 | respawnJoinBound = 100 * time.Millisecond |
| 139 | t.Cleanup(func() { respawnJoinBound = old }) |
| 140 | } |
| 141 | a, b, _, _, ver := stagedActivator(t) |
| 142 | a.gens = false |
| 143 | path := release.GenerationsPath(t.TempDir()) |
| 144 | a.GenerationsFile = path |
| 145 | rt := alwaysOnWorkerRoute("bot10.example.com") |
| 146 | if err := a.WakeWorker(rt); err != nil { |
| 147 | t.Fatal(err) |
| 148 | } |
| 149 | old := b.lastWorker() |
| 150 | rep := b.lastRep() |
| 151 | ver.Store("v2") // the respawn (from the live link) execs v2 |
| 152 | gate := make(chan struct{}) |
| 153 | b.mu.Lock() |
| 154 | if blocked { |
| 155 | b.workerStopGate = gate |
| 156 | } else { |
| 157 | b.workerStuck = true |
| 158 | } |
| 159 | b.workerStartHook = func(registry.Route) { rep.setErr(errors.New("renewal failed")) } |
| 160 | b.mu.Unlock() |
| 161 | a.mu.Lock() |
| 162 | e := a.entries[rt.Host] |
| 163 | a.respawnWorkerLocked(e, nil, "test: shed after the exec", nil) |
| 164 | done := e.respawnDone |
| 165 | a.mu.Unlock() |
| 166 | var fresh *fakeProc |
| 167 | waitFor(t, "the respawn's exec", func() bool { fresh = b.lastWorker(); return fresh != old }) |
| 168 | if blocked { |
| 169 | waitFor(t, "the shed Stop to block", func() bool { return fresh.stopEntered() > 0 }) |
| 170 | } else { |
| 171 | <-done |
| 172 | } |
| 173 | if fresh.Exited() { |
| 174 | t.Fatal("fixture: the shed child should still be running") |
| 175 | } |
| 176 | if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") { |
| 177 | t.Errorf("generations file lists %v while the shed v2 child may still run; want v2 kept", got) |
| 178 | } |
| 179 | // The lost lease begins the terminal stop. |
| 180 | a.sweepOnce() |
| 181 | waitFor(t, "the stop to finish or be held", func() bool { |
| 182 | return rep.isClosed() || !a.entryPresent(rt.Host) || a.entryPendingStop(rt.Host) |
| 183 | }) |
| 184 | if rep.isClosed() || !a.entryPresent(rt.Host) { |
| 185 | t.Fatalf("the stop released the lease (closed %v) and forgot the entry (present %v) while the shed child still runs", rep.isClosed(), a.entryPresent(rt.Host)) |
| 186 | } |
| 187 | if !old.isStopped() { |
| 188 | t.Fatal("the stop did not stop the installed child") |
| 189 | } |
| 190 | if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") { |
| 191 | t.Fatalf("generations file lists %v during the held stop; want v2 kept", got) |
| 192 | } |
| 193 | // Proof: the shed child exits. The stop then finishes, and the |
| 194 | // build is unpinned. |
| 195 | if blocked { |
| 196 | close(gate) |
| 197 | } else { |
| 198 | fresh.exit() |
| 199 | } |
| 200 | waitFor(t, "the stop to finish once the shed child is proven gone", func() bool { |
| 201 | a.sweepOnce() |
| 202 | return rep.isClosed() && !a.entryPresent(rt.Host) |
| 203 | }) |
| 204 | if got := generationsHost(t, a, path, rt.Host); slices.Contains(got, "v2") { |
| 205 | t.Fatalf("generations file lists %v after the shed child's exit; want v2 dropped", got) |
| 206 | } |
| 207 | }) |
| 208 | } |
| 209 | } |
| 210 | |
amadan