| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "strings" |
| 12 | "sync" |
| 13 | "sync/atomic" |
| 14 | "testing" |
| 15 | "time" |
| 16 | |
| 17 | "github.com/carlosframework/platform/internal/cgroup" |
| 18 | "github.com/carlosframework/platform/internal/registry" |
| 19 | ) |
| 20 | |
| 21 | // fakeSwapBackend is fakeBackend plus the swap seam (SwapBackend). Every call is recorded in |
| 22 | // the shared order log: raise:<incumbent>+<successor>, startGen:<path>, |
| 23 | // switch:<n>, restore:<tier>, |
| 24 | // startSidecarGen:<path>. Generation children are fakeProcs labelled |
| 25 | // gen:<n>. |
| 26 | type fakeSwapBackend struct { |
| 27 | *fakeBackend |
| 28 | gmu sync.Mutex |
| 29 | genCalls int |
| 30 | gens []Gen |
| 31 | genProcs []*fakeProc |
| 32 | genErr error |
| 33 | genGate chan struct{} // when set, StartGen blocks until it closes or ctx ends |
| 34 | // genHold, when set, makes StartStaged wait for it to close IGNORING ctx, |
| 35 | // then return a live successor even though the attempt was cancelled: a |
| 36 | // spawn that outlives cancellation, which the barrier must wait for. |
| 37 | genHold chan struct{} |
| 38 | raises int |
| 39 | restores int |
| 40 | // genSocketDir, when set, puts generation sockets there (Task 12), and |
| 41 | // genServe binds a real server on each one, stopped with the process. |
| 42 | genSocketDir string |
| 43 | genServe func(g Gen) (stop func()) |
| 44 | // Task 7: legacy sidecars started from a generation. |
| 45 | sidecarGens []Gen |
| 46 | sidecarGenGate chan struct{} |
| 47 | // sidecarStopGate, when set, is installed on every sidecar started from |
| 48 | // then on: its Stop blocks until the gate closes. |
| 49 | sidecarStopGate chan struct{} |
| 50 | // sidecarDescendant: a process a sidecar forked is still in its cgroup, |
| 51 | // so KillSidecar (Task 7) cannot prove it empty. |
| 52 | sidecarDescendant bool |
| 53 | // sidecarKills counts KillSidecar calls: a test waits on it to know a |
| 54 | // proof was attempted before it asserts nothing was released. |
| 55 | sidecarKills int |
| 56 | // tiers is each build's own tier (GenTier); a build not here is the |
| 57 | // default tier. |
| 58 | tiers map[string]cgroup.Limit |
| 59 | // onRaise, when set, runs inside RaiseGenParent: a test's window between |
| 60 | // an attempt's admission and its exec. |
| 61 | onRaise func() |
| 62 | // genStuck makes every generation spawned from then on a stuck |
| 63 | // fakeProc: its Stop answers ErrStillRunning until exit(). |
| 64 | genStuck bool |
| 65 | // raiseErr is what RaiseGenParent answers after recording the call: a |
| 66 | // raise whose first write landed and a later one failed. |
| 67 | raiseErr error |
| 68 | // restoreFails is how many more RestoreGenParent calls fail. |
| 69 | restoreFails int |
| 70 | // restoreGate, when set, holds every RestoreGenParent call before its |
| 71 | // write until it closes; restoreEntered, when set, is sent to |
| 72 | // (non-blocking) as each call reaches the gate. A held call is counted |
| 73 | // and recorded only once it writes. |
| 74 | restoreGate chan struct{} |
| 75 | restoreEntered chan struct{} |
| 76 | } |
| 77 | |
| 78 | func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen Gen) (Gen, Proc, error) { |
| 79 | g.record("startGen:" + gen.StagedPath) |
| 80 | g.gmu.Lock() |
| 81 | g.genCalls++ |
| 82 | gate, err := g.genGate, g.genErr |
| 83 | g.gmu.Unlock() |
| 84 | if gate != nil { |
| 85 | select { |
| 86 | case <-gate: |
| 87 | case <-ctx.Done(): |
| 88 | return gen, nil, ctx.Err() |
| 89 | } |
| 90 | } |
| 91 | if err != nil { |
| 92 | return gen, nil, err |
| 93 | } |
| 94 | // The caller's exec gate at the launch point (final review F1), after |
| 95 | // the preparation genGate stands for, as the real backend asks it. |
| 96 | if err := runExecGate(ctx); err != nil { |
| 97 | return gen, nil, err |
| 98 | } |
| 99 | g.gmu.Lock() |
| 100 | hold := g.genHold |
| 101 | g.gmu.Unlock() |
| 102 | if hold != nil { |
| 103 | <-hold |
| 104 | } |
| 105 | gen.Socket = fmt.Sprintf("/run/carlos/g/test.%d.sock", gen.N) |
| 106 | if g.genSocketDir != "" { |
| 107 | gen.Socket = filepath.Join(g.genSocketDir, fmt.Sprintf("g%d.sock", gen.N)) |
| 108 | } |
| 109 | gen.Control = gen.Socket + ".control" |
| 110 | g.gmu.Lock() |
| 111 | stuck := g.genStuck |
| 112 | g.gmu.Unlock() |
| 113 | p := &fakeProc{label: fmt.Sprintf("gen:%d", gen.N), rec: g.fakeBackend, stuck: stuck} |
| 114 | var proc Proc = p |
| 115 | if g.genServe != nil { |
| 116 | proc = &genServingProc{fakeProc: p, stop: g.genServe(gen)} |
| 117 | } |
| 118 | g.gmu.Lock() |
| 119 | g.gens = append(g.gens, gen) |
| 120 | g.genProcs = append(g.genProcs, p) |
| 121 | g.gmu.Unlock() |
| 122 | return gen, proc, nil |
| 123 | } |
| 124 | |
| 125 | // GenTier answers from tiers (by build), and the default tier otherwise. |
| 126 | func (g *fakeSwapBackend) GenTier(_ registry.Route, build string) cgroup.Limit { |
| 127 | g.gmu.Lock() |
| 128 | defer g.gmu.Unlock() |
| 129 | if t, ok := g.tiers[build]; ok { |
| 130 | return t |
| 131 | } |
| 132 | return cgroup.TierDefault |
| 133 | } |
| 134 | |
| 135 | func (g *fakeSwapBackend) GenReady(registry.Route, Gen, Proc) {} |
| 136 | |
| 137 | func (g *fakeSwapBackend) RaiseGenParent(_ context.Context, _ registry.Route, inc, succ cgroup.Limit) error { |
| 138 | g.record(fmt.Sprintf("raise:%d+%d", inc.MemoryMax, succ.MemoryMax)) |
| 139 | g.gmu.Lock() |
| 140 | g.raises++ |
| 141 | hook, err := g.onRaise, g.raiseErr |
| 142 | g.gmu.Unlock() |
| 143 | if hook != nil { |
| 144 | hook() |
| 145 | } |
| 146 | return err |
| 147 | } |
| 148 | |
| 149 | func (g *fakeSwapBackend) RestoreGenParent(_ context.Context, _ registry.Route, tier cgroup.Limit) error { |
| 150 | g.gmu.Lock() |
| 151 | gate, entered := g.restoreGate, g.restoreEntered |
| 152 | g.gmu.Unlock() |
| 153 | if gate != nil { |
| 154 | if entered != nil { |
| 155 | select { |
| 156 | case entered <- struct{}{}: |
| 157 | default: |
| 158 | } |
| 159 | } |
| 160 | <-gate |
| 161 | } |
| 162 | g.record(fmt.Sprintf("restore:%d", tier.MemoryMax)) |
| 163 | g.gmu.Lock() |
| 164 | defer g.gmu.Unlock() |
| 165 | g.restores++ |
| 166 | if g.restoreFails > 0 { |
| 167 | g.restoreFails-- |
| 168 | return errors.New("cgroup: parent memory.max: device or resource busy") |
| 169 | } |
| 170 | return nil |
| 171 | } |
| 172 | |
| 173 | func (g *fakeSwapBackend) GenSwitched(_ registry.Route, gen Gen, _ Proc) error { |
| 174 | g.record(fmt.Sprintf("switch:%d", gen.N)) |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | func (g *fakeSwapBackend) calls() (gens, raises, restores int) { |
| 179 | g.gmu.Lock() |
| 180 | defer g.gmu.Unlock() |
| 181 | return g.genCalls, g.raises, g.restores |
| 182 | } |
| 183 | |
| 184 | func (g *fakeSwapBackend) lastGen() (Gen, *fakeProc) { |
| 185 | g.gmu.Lock() |
| 186 | defer g.gmu.Unlock() |
| 187 | if len(g.gens) == 0 { |
| 188 | return Gen{}, nil |
| 189 | } |
| 190 | return g.gens[len(g.gens)-1], g.genProcs[len(g.genProcs)-1] |
| 191 | } |
| 192 | |
| 193 | // genServingProc is a generation child with a real socket server behind it. |
| 194 | // Plan 1 Task 4 already declares a servingProc test type in this package |
| 195 | // (canonlink_test.go, wrapping a Proc and an *http.Server), so this one has |
| 196 | // its own name. |
| 197 | type genServingProc struct { |
| 198 | *fakeProc |
| 199 | stop func() |
| 200 | } |
| 201 | |
| 202 | func (p *genServingProc) Stop(ctx context.Context) error { |
| 203 | p.stop() |
| 204 | return p.fakeProc.Stop(ctx) |
| 205 | } |
| 206 | |
| 207 | // swapFixture is an awake host on a swap-enabled Service whose backend runs |
| 208 | // generations. The incumbent is generation 1<<40 on build v1, a number |
| 209 | // nextGen will never hand out in a test. Build v2 is staged on disk with |
| 210 | // its real digest. |
| 211 | type swapFixture struct { |
| 212 | a *Service |
| 213 | b *fakeSwapBackend |
| 214 | clk *fakeClock |
| 215 | reg *registry.Registry |
| 216 | host string |
| 217 | incumbent Gen |
| 218 | staged string |
| 219 | digest string |
| 220 | rev atomic.Uint64 |
| 221 | } |
| 222 | |
| 223 | func newSwapFixture(t *testing.T) *swapFixture { |
| 224 | t.Helper() |
| 225 | return newSwapFixtureOn(t, nil) |
| 226 | } |
| 227 | |
| 228 | // newSwapFixtureOn is newSwapFixture with the Service's backend made by |
| 229 | // wrap from the swap fake, for a test that needs one more capability. |
| 230 | func newSwapFixtureOn(t *testing.T, wrap func(*fakeSwapBackend) Backend) *swapFixture { |
| 231 | t.Helper() |
| 232 | b := &fakeSwapBackend{fakeBackend: &fakeBackend{}} |
| 233 | clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| 234 | reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| 235 | if err != nil { |
| 236 | t.Fatal(err) |
| 237 | } |
| 238 | var be Backend = b |
| 239 | if wrap != nil { |
| 240 | be = wrap(b) |
| 241 | } |
| 242 | a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, be, reg) |
| 243 | a.gens = true // plan 1's switch, as plan 5 sets it from CARLOS_SWAP |
| 244 | a.now = clk.now |
| 245 | a.genHealth = func(context.Context, Gen) error { return nil } |
| 246 | f := &swapFixture{a: a, b: b, clk: clk, reg: reg, host: "swap.example.com"} |
| 247 | // The route's socket and database live in the test's own directory, |
| 248 | // never the shared /tmp. |
| 249 | rt := testRoute(f.host) |
| 250 | rdir := t.TempDir() |
| 251 | rt.Socket, rt.DBPath = filepath.Join(rdir, "app.sock"), filepath.Join(rdir, "app.db") |
| 252 | // Generation sockets too: a hold's pinned wake runs wake()'s awake |
| 253 | // transition, which points the canonical path at the generation's socket |
| 254 | // for real (swapCanonicalLink). |
| 255 | b.genSocketDir = t.TempDir() |
| 256 | arrangeAwake(t, a, rt) |
| 257 | f.incumbent = Gen{N: 1 << 40, Socket: "/run/carlos/g/incumbent.sock", Control: "/run/carlos/g/incumbent.sock.control", |
| 258 | Build: "v1", StagedPath: "/opt/carlos/versions/v1/app"} |
| 259 | setCurrentGen(t, a, f.host, f.incumbent) |
| 260 | dir := filepath.Join(t.TempDir(), "versions", "v2") |
| 261 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | f.staged = filepath.Join(dir, "app") |
| 265 | if err := os.WriteFile(f.staged, []byte("build v2"), 0o755); err != nil { |
| 266 | t.Fatal(err) |
| 267 | } |
| 268 | sum := sha256.Sum256([]byte("build v2")) |
| 269 | f.digest = hex.EncodeToString(sum[:]) |
| 270 | f.rev.Store(1) |
| 271 | return f |
| 272 | } |
| 273 | |
| 274 | func (f *swapFixture) attempt(mode SwapMode) Attempt { |
| 275 | return Attempt{Host: f.host, Cause: CauseRelease, Build: "v2", StagedPath: f.staged, Digest: f.digest, Mode: mode, SchedRev: f.rev.Load()} |
| 276 | } |
| 277 | |
| 278 | func (f *swapFixture) newest(string) uint64 { return f.rev.Load() } |
| 279 | |
| 280 | func (f *swapFixture) appSocket(t *testing.T) string { |
| 281 | t.Helper() |
| 282 | sock, ok := f.a.AppSocket(f.host) |
| 283 | if !ok { |
| 284 | t.Fatalf("AppSocket(%s) reported no current generation", f.host) |
| 285 | } |
| 286 | return sock |
| 287 | } |
| 288 | |
| 289 | // setCurrentGen stands in for plan 1's wake, which records the generation |
| 290 | // it spawned in e.gen (contract C2, C3). |
| 291 | func setCurrentGen(t *testing.T, a *Service, host string, g Gen) { |
| 292 | t.Helper() |
| 293 | a.mu.Lock() |
| 294 | defer a.mu.Unlock() |
| 295 | e := a.entries[host] |
| 296 | if e == nil || e.state != actAwake { |
| 297 | t.Fatalf("%s is not awake", host) |
| 298 | } |
| 299 | e.gen, e.spawnVersion = g, g.Build |
| 300 | } |
| 301 | |
| 302 | // genStateForTest is the entry's current generation and its operation set |
| 303 | // by role. |
| 304 | func (a *Service) genStateForTest(host string) (Gen, map[uint64]genRole) { |
| 305 | a.mu.Lock() |
| 306 | defer a.mu.Unlock() |
| 307 | e := a.entries[host] |
| 308 | if e == nil { |
| 309 | return Gen{}, nil |
| 310 | } |
| 311 | out := make(map[uint64]genRole, len(e.ops.members)) |
| 312 | for n, m := range e.ops.members { |
| 313 | out[n] = m.role |
| 314 | } |
| 315 | return e.gen, out |
| 316 | } |
| 317 | |
| 318 | // indexPrefix is slices.Index for the first event with prefix p. |
| 319 | func indexPrefix(ev []string, p string) int { |
| 320 | for i, e := range ev { |
| 321 | if strings.HasPrefix(e, p) { |
| 322 | return i |
| 323 | } |
| 324 | } |
| 325 | return -1 |
| 326 | } |
| 327 | |
| 328 | func (g *fakeSwapBackend) StartSidecarGen(_ context.Context, _ registry.Route, gen Gen) (Proc, error) { |
| 329 | g.record("startSidecarGen:" + gen.StagedPath) |
| 330 | g.gmu.Lock() |
| 331 | g.sidecarGens = append(g.sidecarGens, gen) |
| 332 | gate := g.sidecarGenGate |
| 333 | g.gmu.Unlock() |
| 334 | if gate != nil { |
| 335 | <-gate |
| 336 | } |
| 337 | g.gmu.Lock() |
| 338 | stopGate := g.sidecarStopGate |
| 339 | g.gmu.Unlock() |
| 340 | p := &fakeProc{label: "sidecar", rec: g.fakeBackend, stopGate: stopGate} |
| 341 | g.mu.Lock() |
| 342 | g.sidecars = append(g.sidecars, p) |
| 343 | g.mu.Unlock() |
| 344 | return p, nil |
| 345 | } |
| 346 | |
| 347 | // KillSidecar is the sidecar cgroup's proof: it fails while a descendant |
| 348 | // is "left" (setSidecarDescendant). |
| 349 | func (g *fakeSwapBackend) KillSidecar(context.Context, registry.Route) error { |
| 350 | g.gmu.Lock() |
| 351 | defer g.gmu.Unlock() |
| 352 | g.sidecarKills++ |
| 353 | if g.sidecarDescendant { |
| 354 | return errors.New("role-sidecar: still populated") |
| 355 | } |
| 356 | return nil |
| 357 | } |
| 358 | |
| 359 | func (g *fakeSwapBackend) setSidecarDescendant(left bool) { |
| 360 | g.gmu.Lock() |
| 361 | g.sidecarDescendant = left |
| 362 | g.gmu.Unlock() |
| 363 | } |
| 364 | |
| 365 | func (g *fakeSwapBackend) sidecarKillCount() int { |
| 366 | g.gmu.Lock() |
| 367 | defer g.gmu.Unlock() |
| 368 | return g.sidecarKills |
| 369 | } |
| 370 | |
| 371 | func (g *fakeSwapBackend) sidecarGenPaths() []string { |
| 372 | g.gmu.Lock() |
| 373 | defer g.gmu.Unlock() |
| 374 | out := make([]string, len(g.sidecarGens)) |
| 375 | for i, s := range g.sidecarGens { |
| 376 | out[i] = s.StagedPath |
| 377 | } |
| 378 | return out |
| 379 | } |
| 380 | |
amadan