| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "crypto/sha256" |
| 6 | "encoding/hex" |
| 7 | "errors" |
| 8 | "fmt" |
| 9 | "log" |
| 10 | "math/rand" |
| 11 | "os" |
| 12 | "path/filepath" |
| 13 | "regexp" |
| 14 | "sort" |
| 15 | "strconv" |
| 16 | "strings" |
| 17 | "sync" |
| 18 | "sync/atomic" |
| 19 | "testing" |
| 20 | "time" |
| 21 | |
| 22 | "github.com/carlosframework/platform/internal/cgroup" |
| 23 | "github.com/carlosframework/platform/internal/registry" |
| 24 | ) |
| 25 | |
| 26 | // TestSwapInvariants drives random schedules of requests, overlap and hold |
| 27 | // swaps, hibernation, lease loss, restarts, maintenance fences, root |
| 28 | // repoints, supersession, crashes, re-stop cadences and a final Shutdown |
| 29 | // against the real Service and a simulated backend. It checks I1, I2 and I3 |
| 30 | // (plan 3 task 13; spec §12). |
| 31 | // |
| 32 | // The simulated processes do not all stop when asked. Some drain slowly, |
| 33 | // some answer ErrStillRunning or a failed kill and exit on their own later, |
| 34 | // some generations fork a process that outlives its leader, KillGen can |
| 35 | // fail before it heals, and a start can fail with its kill unproven |
| 36 | // (errGenUnproven), with or without a process handle. So the |
| 37 | // retain-until-proven paths (pending stops, the member and wake reapers) |
| 38 | // run under the sweep too. Each seed also picks whether generations are on |
| 39 | // (the dormant switch) and whether generation containment is. |
| 40 | // |
| 41 | // Seeds: SWAP_SIM_SEEDS (default 100). SWAP_SIM_SEED=<n> replays one seed. |
| 42 | // Goroutine interleaving is not deterministic, so a replay is best effort. |
| 43 | // The event log printed with a failure is the evidence. |
| 44 | // |
| 45 | // The mutation checks (testdata/swapsim/mutate.sh) run this sweep and the |
| 46 | // TestSwapWitness* tests against each `// swapsim:Mn` mutant. |
| 47 | func TestSwapInvariants(t *testing.T) { |
| 48 | first, last := 0, 100 |
| 49 | if v := os.Getenv("SWAP_SIM_SEEDS"); v != "" { |
| 50 | n, err := strconv.Atoi(v) |
| 51 | if err != nil { |
| 52 | t.Fatal(err) |
| 53 | } |
| 54 | last = n |
| 55 | } |
| 56 | if v := os.Getenv("SWAP_SIM_SEED"); v != "" { |
| 57 | n, err := strconv.Atoi(v) |
| 58 | if err != nil { |
| 59 | t.Fatal(err) |
| 60 | } |
| 61 | first, last = n, n+1 |
| 62 | } |
| 63 | oldSettle, oldJoin := swapSettle, opJoinBound |
| 64 | swapSettle, opJoinBound = 2*time.Millisecond, 3*time.Second |
| 65 | t.Cleanup(func() { swapSettle, opJoinBound = oldSettle, oldJoin }) |
| 66 | // The Service's own log lines go into the seed's event log, so a failure |
| 67 | // prints them in order with the model's events, and a passing sweep |
| 68 | // prints nothing. |
| 69 | sink := &simLogSink{} |
| 70 | oldOut, oldFlags := log.Writer(), log.Flags() |
| 71 | log.SetOutput(sink) |
| 72 | log.SetFlags(0) |
| 73 | t.Cleanup(func() { log.SetOutput(oldOut); log.SetFlags(oldFlags) }) |
| 74 | staged := stageSimBuilds(t) |
| 75 | sockDir := t.TempDir() |
| 76 | failed := 0 |
| 77 | tally := map[string]int{} |
| 78 | t.Cleanup(func() { |
| 79 | keys := make([]string, 0, len(tally)) |
| 80 | for k := range tally { |
| 81 | keys = append(keys, k) |
| 82 | } |
| 83 | sort.Strings(keys) |
| 84 | lines := make([]string, len(keys)) |
| 85 | for i, k := range keys { |
| 86 | lines[i] = fmt.Sprintf("%6d %s", tally[k], k) |
| 87 | } |
| 88 | t.Logf("seeds %d..%d: seed settings and Swap results (mode/outcome: reason, normalized):\n%s", first, last-1, strings.Join(lines, "\n")) |
| 89 | }) |
| 90 | for seed := first; seed < last; seed++ { |
| 91 | violations, events, outcomes := simulateSwapOnce(seed, staged, sockDir, sink) |
| 92 | for k, n := range outcomes { |
| 93 | tally[k] += n |
| 94 | } |
| 95 | if len(violations) == 0 { |
| 96 | continue |
| 97 | } |
| 98 | failed++ |
| 99 | if len(events) > 200 { |
| 100 | events = events[len(events)-200:] |
| 101 | } |
| 102 | // A seed whose only failure is liveness (a Swap that never returned) |
| 103 | // says so: it is not an invariant violation. |
| 104 | kind := "liveness failed" |
| 105 | for _, v := range violations { |
| 106 | if !strings.HasPrefix(v, "liveness:") { |
| 107 | kind = "invariant violated" |
| 108 | } |
| 109 | } |
| 110 | t.Errorf("%s at seed %d (replay: SWAP_SIM_SEED=%d): %s\n--- last events ---\n%s", kind, seed, seed, strings.Join(violations, "; "), strings.Join(events, "\n")) |
| 111 | if failed >= 3 { |
| 112 | t.FailNow() |
| 113 | } |
| 114 | } |
| 115 | } |
| 116 | |
| 117 | type simArtifact struct{ path, digest string } |
| 118 | |
| 119 | func stageSimBuilds(t *testing.T) map[string]simArtifact { |
| 120 | t.Helper() |
| 121 | out := map[string]simArtifact{} |
| 122 | for _, b := range []string{"v3", "v4"} { |
| 123 | dir := filepath.Join(t.TempDir(), "versions", b) |
| 124 | if err := os.MkdirAll(dir, 0o755); err != nil { |
| 125 | t.Fatal(err) |
| 126 | } |
| 127 | p := filepath.Join(dir, "app") |
| 128 | body := []byte("build " + b) |
| 129 | if err := os.WriteFile(p, body, 0o755); err != nil { |
| 130 | t.Fatal(err) |
| 131 | } |
| 132 | sum := sha256.Sum256(body) |
| 133 | out[b] = simArtifact{p, hex.EncodeToString(sum[:])} |
| 134 | } |
| 135 | return out |
| 136 | } |
| 137 | |
| 138 | // simLogSink routes the log package into the current seed's world. A |
| 139 | // goroutine a previous seed left behind may log into the next seed's |
| 140 | // events; the lines say what they are. |
| 141 | type simLogSink struct{ w atomic.Pointer[simWorld] } |
| 142 | |
| 143 | func (s *simLogSink) Write(p []byte) (int, error) { |
| 144 | if w := s.w.Load(); w != nil { |
| 145 | w.logf("log: %s", strings.TrimRight(string(p), "\n")) |
| 146 | } |
| 147 | return len(p), nil |
| 148 | } |
| 149 | |
| 150 | // simWorld is the model: which simulated processes run on each host, the |
| 151 | // live link, the staged builds, and the violations seen. Lock order: a |
| 152 | // caller may hold a.mu when it takes w.mu (the backend is called under the |
| 153 | // entry lock), never the reverse; a process's own p.mu nests inside w.mu |
| 154 | // (die), never the reverse. |
| 155 | type simWorld struct { |
| 156 | mu sync.Mutex |
| 157 | rng *rand.Rand |
| 158 | events []string |
| 159 | running map[string]map[*simProc]bool |
| 160 | live map[string]string |
| 161 | staged map[string]string |
| 162 | violations []string |
| 163 | seen map[string]bool |
| 164 | nextID int |
| 165 | hosts []string |
| 166 | // gensOn is this seed's dormant switch (Service.gens). contained is |
| 167 | // generation containment: with it, each generation has its own child |
| 168 | // that KillGen empties, and a start can fail with its kill unproven; |
| 169 | // without it, KillGen answers errNoGenContainment, the parent cannot be |
| 170 | // raised, and a generation's leader is all there is of it. |
| 171 | gensOn, contained bool |
| 172 | // killFails is how many more KillGen calls fail for host#n before the |
| 173 | // kill heals. |
| 174 | killFails map[string]int |
| 175 | // shutdown is set once Service.Shutdown has returned: nothing may spawn |
| 176 | // after it (I2). |
| 177 | shutdown bool |
| 178 | // outcomes counts the seed's settings and its Swap results by mode, |
| 179 | // outcome and reason, so a sweep shows what it reached. |
| 180 | outcomes map[string]int |
| 181 | // dying, when set, runs once a death is published, with no model lock |
| 182 | // held: the moment a reaper that saw Exited() could act. The model's |
| 183 | // own interleaving test sets it; the sweep never does. |
| 184 | dying func(p *simProc) |
| 185 | } |
| 186 | |
| 187 | func (w *simWorld) logf(format string, args ...any) { |
| 188 | w.mu.Lock() |
| 189 | w.events = append(w.events, fmt.Sprintf(format, args...)) |
| 190 | w.mu.Unlock() |
| 191 | } |
| 192 | |
| 193 | func (w *simWorld) violatef(format string, args ...any) { |
| 194 | msg := fmt.Sprintf(format, args...) |
| 195 | w.mu.Lock() |
| 196 | defer w.mu.Unlock() |
| 197 | if w.seen[msg] { |
| 198 | return |
| 199 | } |
| 200 | w.seen[msg] = true |
| 201 | w.violations = append(w.violations, msg) |
| 202 | w.events = append(w.events, "VIOLATION "+msg) |
| 203 | } |
| 204 | |
| 205 | func (w *simWorld) intn(n int) int { |
| 206 | w.mu.Lock() |
| 207 | defer w.mu.Unlock() |
| 208 | return w.rng.Intn(n) |
| 209 | } |
| 210 | |
| 211 | func (w *simWorld) jitter() { |
| 212 | if d := w.intn(4); d > 0 { |
| 213 | time.Sleep(time.Duration(d) * 250 * time.Microsecond) |
| 214 | } |
| 215 | } |
| 216 | |
| 217 | // spawn starts one simulated process of generation n. A leader gets a |
| 218 | // random stop behaviour; a descendant (a process the app forked) stops |
| 219 | // only when killed. |
| 220 | func (w *simWorld) spawn(host, build string, n uint64, leader bool) *simProc { |
| 221 | w.mu.Lock() |
| 222 | w.nextID++ |
| 223 | p := &simProc{w: w, id: w.nextID, host: host, build: build, n: n, leader: leader} |
| 224 | if leader { |
| 225 | switch r := w.rng.Intn(20); { |
| 226 | case r < 2: |
| 227 | p.mode = procStubborn |
| 228 | case r < 3: |
| 229 | // A failed kill that is not ErrStillRunning is a claims-on |
| 230 | // process's answer (KillAndEmpty), and the generations-on stop |
| 231 | // handles it through releasable. With generations off the exec |
| 232 | // backends' Stop (childProc, unitProc) answers ErrStillRunning |
| 233 | // for a process that may remain, so that is what it gets. |
| 234 | p.mode = procStubborn |
| 235 | if w.gensOn { |
| 236 | p.mode = procKillFails |
| 237 | } |
| 238 | case r < 7: |
| 239 | p.mode = procSlow |
| 240 | } |
| 241 | } |
| 242 | if w.running[host] == nil { |
| 243 | w.running[host] = map[*simProc]bool{} |
| 244 | } |
| 245 | w.running[host][p] = true |
| 246 | what := "spawn" |
| 247 | if !leader { |
| 248 | what = "fork" |
| 249 | } |
| 250 | w.events = append(w.events, fmt.Sprintf("%s p%d %s build=%s gen=%d mode=%s", what, p.id, host, build, n, p.mode)) |
| 251 | after := w.shutdown && w.gensOn |
| 252 | w.mu.Unlock() |
| 253 | if after { |
| 254 | w.violatef("I2: p%d (%s build %s) spawned after Shutdown returned", p.id, host, build) |
| 255 | } |
| 256 | return p |
| 257 | } |
| 258 | |
| 259 | // spawnGen starts generation n's leader. With generations and containment |
| 260 | // on, one generation in four also forks a process that outlives its |
| 261 | // leader, which only KillGen (or a crash) ends. |
| 262 | func (w *simWorld) spawnGen(host, build string, n uint64) *simProc { |
| 263 | p := w.spawn(host, build, n, true) |
| 264 | if w.gensOn && w.contained && n != 0 && w.intn(4) == 0 { |
| 265 | w.spawn(host, build, n, false) |
| 266 | } |
| 267 | return p |
| 268 | } |
| 269 | |
| 270 | func (w *simWorld) runningOn(host string) int { |
| 271 | w.mu.Lock() |
| 272 | defer w.mu.Unlock() |
| 273 | return len(w.running[host]) |
| 274 | } |
| 275 | |
| 276 | func (w *simWorld) runningCount() int { |
| 277 | w.mu.Lock() |
| 278 | defer w.mu.Unlock() |
| 279 | n := 0 |
| 280 | for _, ps := range w.running { |
| 281 | n += len(ps) |
| 282 | } |
| 283 | return n |
| 284 | } |
| 285 | |
| 286 | func (w *simWorld) leaseFreed(host, by string) { |
| 287 | if n := w.runningOn(host); n > 0 { |
| 288 | w.violatef("I2: lease of %s released by %s while %d generation process(es) run", host, by, n) |
| 289 | } |
| 290 | w.logf("lease %s released by %s", host, by) |
| 291 | } |
| 292 | |
| 293 | func (w *simWorld) setLive(host, build string) { |
| 294 | w.mu.Lock() |
| 295 | w.live[host] = build |
| 296 | w.events = append(w.events, "repoint "+host+" -> "+build) |
| 297 | w.mu.Unlock() |
| 298 | } |
| 299 | |
| 300 | func (w *simWorld) crashOne(host string) { |
| 301 | w.mu.Lock() |
| 302 | var victims []*simProc |
| 303 | for p := range w.running[host] { |
| 304 | victims = append(victims, p) |
| 305 | } |
| 306 | // Map order is random; sort so the seed picks the victim. |
| 307 | sort.Slice(victims, func(i, j int) bool { return victims[i].id < victims[j].id }) |
| 308 | var p *simProc |
| 309 | if len(victims) > 0 { |
| 310 | p = victims[w.rng.Intn(len(victims))] |
| 311 | } |
| 312 | w.mu.Unlock() |
| 313 | if p != nil { |
| 314 | p.die("crash") |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | var simDigits = regexp.MustCompile(`[0-9]+`) |
| 319 | |
| 320 | // tally counts one Swap result. The reason is normalized (host names and |
| 321 | // numbers out, length capped) so that like results group. |
| 322 | func (w *simWorld) tally(res SwapResult) { |
| 323 | key := string(res.Mode) + "/" + string(res.Outcome) |
| 324 | if res.Outcome != SwapDone && res.Reason != "" { |
| 325 | r := res.Reason |
| 326 | for _, h := range w.hosts { |
| 327 | r = strings.ReplaceAll(r, h, "<host>") |
| 328 | } |
| 329 | r = simDigits.ReplaceAllString(r, "N") |
| 330 | if len(r) > 110 { |
| 331 | r = r[:110] + "..." |
| 332 | } |
| 333 | key += ": " + r |
| 334 | } |
| 335 | w.mu.Lock() |
| 336 | w.outcomes[key]++ |
| 337 | w.mu.Unlock() |
| 338 | } |
| 339 | |
| 340 | func (w *simWorld) snapshot() ([]string, []string, map[string]int) { |
| 341 | w.mu.Lock() |
| 342 | defer w.mu.Unlock() |
| 343 | outcomes := make(map[string]int, len(w.outcomes)) |
| 344 | for k, n := range w.outcomes { |
| 345 | outcomes[k] = n |
| 346 | } |
| 347 | return append([]string(nil), w.violations...), append([]string(nil), w.events...), outcomes |
| 348 | } |
| 349 | |
| 350 | // simProcMode is how a simulated leader answers Stop. |
| 351 | type simProcMode int |
| 352 | |
| 353 | const ( |
| 354 | // procPrompt ends at once. |
| 355 | procPrompt simProcMode = iota |
| 356 | // procSlow drains for a few milliseconds, then ends. |
| 357 | procSlow |
| 358 | // procStubborn answers ErrStillRunning: SIGKILL did not prove it gone. |
| 359 | // It exits on its own a few milliseconds after it was first asked. |
| 360 | procStubborn |
| 361 | // procKillFails answers a kill error that is not ErrStillRunning (a |
| 362 | // claims-on KillAndEmpty failure), and exits on its own later, as |
| 363 | // procStubborn does. Generations-on seeds only (see spawn). |
| 364 | procKillFails |
| 365 | ) |
| 366 | |
| 367 | func (m simProcMode) String() string { |
| 368 | return [...]string{"prompt", "slow", "stubborn", "killfails"}[m] |
| 369 | } |
| 370 | |
| 371 | var errSimKillFailed = errors.New("sim: kill failed") |
| 372 | |
| 373 | // simProc is one simulated process of a generation: its leader, or a |
| 374 | // process the leader forked (leader false), which nothing but KillGen or a |
| 375 | // crash ends. |
| 376 | type simProc struct { |
| 377 | w *simWorld |
| 378 | id int |
| 379 | host string |
| 380 | build string |
| 381 | n uint64 |
| 382 | leader bool |
| 383 | mode simProcMode |
| 384 | mu sync.Mutex |
| 385 | dead bool |
| 386 | later bool |
| 387 | } |
| 388 | |
| 389 | func (p *simProc) Stop(context.Context) error { |
| 390 | p.w.jitter() |
| 391 | if p.Exited() { |
| 392 | return nil |
| 393 | } |
| 394 | switch p.mode { |
| 395 | case procSlow: |
| 396 | time.Sleep(time.Duration(1+p.w.intn(6)) * time.Millisecond) |
| 397 | case procStubborn: |
| 398 | p.exitLater() |
| 399 | return ErrStillRunning |
| 400 | case procKillFails: |
| 401 | p.exitLater() |
| 402 | return errSimKillFailed |
| 403 | } |
| 404 | p.die("stop") |
| 405 | return nil |
| 406 | } |
| 407 | |
| 408 | // exitLater arms the one delayed exit of a process that did not stop when |
| 409 | // asked. |
| 410 | func (p *simProc) exitLater() { |
| 411 | p.mu.Lock() |
| 412 | armed := p.later |
| 413 | p.later = true |
| 414 | p.mu.Unlock() |
| 415 | if armed { |
| 416 | return |
| 417 | } |
| 418 | d := time.Duration(1+p.w.intn(15)) * time.Millisecond |
| 419 | time.AfterFunc(d, func() { p.die("late exit") }) |
| 420 | } |
| 421 | |
| 422 | func (p *simProc) Exited() bool { |
| 423 | p.mu.Lock() |
| 424 | defer p.mu.Unlock() |
| 425 | return p.dead |
| 426 | } |
| 427 | |
| 428 | // die is one transition of the model: the process is marked dead and |
| 429 | // leaves running in the same critical section (w.mu, then p.mu, the |
| 430 | // model's lock order), so no observer can see it Exited() and still |
| 431 | // running. A concurrent die waits on w.mu and finds it already dead. |
| 432 | func (p *simProc) die(why string) { |
| 433 | p.w.mu.Lock() |
| 434 | p.mu.Lock() |
| 435 | was := p.dead |
| 436 | p.dead = true |
| 437 | p.mu.Unlock() |
| 438 | if !was { |
| 439 | delete(p.w.running[p.host], p) |
| 440 | p.w.events = append(p.w.events, fmt.Sprintf("exit p%d (%s)", p.id, why)) |
| 441 | } |
| 442 | p.w.mu.Unlock() |
| 443 | if !was && p.w.dying != nil { |
| 444 | p.w.dying(p) |
| 445 | } |
| 446 | } |
| 447 | |
| 448 | type simRep struct { |
| 449 | w *simWorld |
| 450 | host string |
| 451 | mu sync.Mutex |
| 452 | err error |
| 453 | } |
| 454 | |
| 455 | func (r *simRep) Err() error { |
| 456 | r.mu.Lock() |
| 457 | defer r.mu.Unlock() |
| 458 | return r.err |
| 459 | } |
| 460 | |
| 461 | func (r *simRep) Close(context.Context) error { |
| 462 | r.w.jitter() |
| 463 | r.w.leaseFreed(r.host, "rep.Close") |
| 464 | return nil |
| 465 | } |
| 466 | |
| 467 | type simLease struct { |
| 468 | b *simBackend |
| 469 | host string |
| 470 | } |
| 471 | |
| 472 | func (l *simLease) Replicate(context.Context, string, time.Duration) (Rep, error) { |
| 473 | r := &simRep{w: l.b.w, host: l.host} |
| 474 | l.b.mu.Lock() |
| 475 | l.b.reps[l.host] = r |
| 476 | l.b.mu.Unlock() |
| 477 | return r, nil |
| 478 | } |
| 479 | |
| 480 | func (l *simLease) Release(context.Context) error { |
| 481 | l.b.w.leaseFreed(l.host, "lease.Release") |
| 482 | return nil |
| 483 | } |
| 484 | |
| 485 | // simBackend is Backend, GenBackend, SwapBackend and genKiller, as the |
| 486 | // exec backend is. It has no hostKiller or sidecarKiller (no workers, no |
| 487 | // sidecars). The live-link wake goes through StartGen, so every generation |
| 488 | // has a number KillGen can name. |
| 489 | type simBackend struct { |
| 490 | w *simWorld |
| 491 | dir string |
| 492 | mu sync.Mutex |
| 493 | reps map[string]*simRep |
| 494 | } |
| 495 | |
| 496 | var ( |
| 497 | _ Backend = (*simBackend)(nil) |
| 498 | _ GenBackend = (*simBackend)(nil) |
| 499 | _ SwapBackend = (*simBackend)(nil) |
| 500 | _ genKiller = (*simBackend)(nil) |
| 501 | _ DueStore = simDue{} |
| 502 | ) |
| 503 | |
| 504 | func (b *simBackend) loseLease(host string) { |
| 505 | b.mu.Lock() |
| 506 | r := b.reps[host] |
| 507 | b.mu.Unlock() |
| 508 | if r != nil { |
| 509 | r.mu.Lock() |
| 510 | r.err = errors.New("sim: lease lost") |
| 511 | r.mu.Unlock() |
| 512 | } |
| 513 | } |
| 514 | |
| 515 | func (b *simBackend) Acquire(_ context.Context, rt registry.Route) (Lease, error) { |
| 516 | b.w.jitter() |
| 517 | if n := b.w.runningOn(rt.Host); n > 0 { |
| 518 | b.w.violatef("I2: lease of %s acquired while %d process(es) of the previous holder's generations run", rt.Host, n) |
| 519 | } |
| 520 | return &simLease{b: b, host: rt.Host}, nil |
| 521 | } |
| 522 | func (b *simBackend) Restore(context.Context, registry.Route) (bool, error) { |
| 523 | b.w.jitter() |
| 524 | return false, nil |
| 525 | } |
| 526 | func (b *simBackend) Seed(context.Context, registry.Route) error { return nil } |
| 527 | |
| 528 | // Start is the pre-generations spawn. The Service never calls it on a |
| 529 | // GenBackend; it is here for the Backend interface. |
| 530 | func (b *simBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) { |
| 531 | p, g, err := b.StartGen(ctx, rt, 0, false) |
| 532 | return p, g.Build, err |
| 533 | } |
| 534 | |
| 535 | // StartGen is the wake's spawn of the live link (which only ever names v1 |
| 536 | // or v2). |
| 537 | func (b *simBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) { |
| 538 | b.w.jitter() |
| 539 | b.w.mu.Lock() |
| 540 | build := b.w.live[rt.Host] |
| 541 | b.w.mu.Unlock() |
| 542 | g := canonicalGen(rt, n, build) |
| 543 | if perGen { |
| 544 | g.Socket = filepath.Join(b.dir, fmt.Sprintf("%s.%d.sock", rt.Host, n)) |
| 545 | g.Control = g.Socket + ".control" |
| 546 | } |
| 547 | if err := ctx.Err(); err != nil { |
| 548 | return nil, g, err |
| 549 | } |
| 550 | p, err := b.startOutcome(rt.Host, build, n) |
| 551 | return p, g, err |
| 552 | } |
| 553 | |
| 554 | // startOutcome is a spawn's result. Most run. Some fail before the exec, |
| 555 | // with nothing running. With containment, some fail after it with their |
| 556 | // kill unproven (errGenUnproven, as plan 2's startUnitAt answers): the |
| 557 | // generation runs on, with its handle returned or not. |
| 558 | func (b *simBackend) startOutcome(host, build string, n uint64) (Proc, error) { |
| 559 | w := b.w |
| 560 | r := w.intn(30) |
| 561 | switch { |
| 562 | case r == 0: |
| 563 | return nil, errors.New("sim: exec failed") |
| 564 | case r <= 2 && w.contained: |
| 565 | p := w.spawnGen(host, build, n) |
| 566 | err := fmt.Errorf("sim: health wait failed; %w: kill unproven", errGenUnproven) |
| 567 | w.logf("start of %s gen %d failed with its kill unproven (handle returned: %v)", host, n, r == 1) |
| 568 | if r == 1 { |
| 569 | return p, err |
| 570 | } |
| 571 | return nil, err |
| 572 | } |
| 573 | return w.spawnGen(host, build, n), nil |
| 574 | } |
| 575 | |
| 576 | func (b *simBackend) NextDueVia(context.Context, registry.Route, string) (int64, error) { |
| 577 | return 0, nil |
| 578 | } |
| 579 | func (b *simBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| 580 | func (b *simBackend) HasSidecar(registry.Route) bool { return false } |
| 581 | func (b *simBackend) StartSidecar(context.Context, registry.Route) (Proc, error) { |
| 582 | return nil, errors.New("sim: no sidecars") |
| 583 | } |
| 584 | func (b *simBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) { |
| 585 | return nil, "", errors.New("sim: no workers") |
| 586 | } |
| 587 | |
| 588 | // StartStaged checks I3 at the exec: the staged path a generation runs is |
| 589 | // the one its build was staged at. Its sockets live in the run's own |
| 590 | // directory, because a hold's pinned wake links the canonical path to the |
| 591 | // generation's socket for real (swapCanonicalLink). |
| 592 | func (b *simBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) { |
| 593 | b.w.mu.Lock() |
| 594 | want := b.w.staged[g.Build] |
| 595 | b.w.mu.Unlock() |
| 596 | if g.StagedPath != want { |
| 597 | b.w.violatef("I3: generation %d of %s execs %q for build %s (staged at %q)", g.N, rt.Host, g.StagedPath, g.Build, want) |
| 598 | } |
| 599 | b.w.jitter() |
| 600 | if err := ctx.Err(); err != nil { |
| 601 | return g, nil, err |
| 602 | } |
| 603 | // The caller's exec gate at the launch point, after the preparation |
| 604 | // the jitter stands for (final review F1). |
| 605 | if err := runExecGate(ctx); err != nil { |
| 606 | return g, nil, err |
| 607 | } |
| 608 | g.Socket = filepath.Join(b.dir, fmt.Sprintf("%s.%d.sock", rt.Host, g.N)) |
| 609 | g.Control = g.Socket + ".control" |
| 610 | p, err := b.startOutcome(rt.Host, g.Build, g.N) |
| 611 | return g, p, err |
| 612 | } |
| 613 | |
| 614 | // KillGen is plan 2's generation kill: it empties generation n's child |
| 615 | // and returns once the child is empty. One generation in six fails its |
| 616 | // first one or two kills, then heals. A process that does not die when |
| 617 | // killed (procStubborn, procKillFails) keeps the child from emptying until |
| 618 | // it exits on its own. Without containment it answers errNoGenContainment, |
| 619 | // as ExecBackend.KillGen does. |
| 620 | func (b *simBackend) KillGen(_ context.Context, rt registry.Route, n uint64) error { |
| 621 | w := b.w |
| 622 | if !w.contained { |
| 623 | return fmt.Errorf("sim: %w", errNoGenContainment) |
| 624 | } |
| 625 | w.jitter() |
| 626 | key := fmt.Sprintf("%s#%d", rt.Host, n) |
| 627 | w.mu.Lock() |
| 628 | fails, seen := w.killFails[key] |
| 629 | if !seen && w.rng.Intn(6) == 0 { |
| 630 | fails = 1 + w.rng.Intn(2) |
| 631 | } |
| 632 | var procs []*simProc |
| 633 | for p := range w.running[rt.Host] { |
| 634 | if p.n == n { |
| 635 | procs = append(procs, p) |
| 636 | } |
| 637 | } |
| 638 | if fails > 0 { |
| 639 | w.killFails[key] = fails - 1 |
| 640 | w.events = append(w.events, fmt.Sprintf("KillGen %s gen %d fails (%d more)", rt.Host, n, fails-1)) |
| 641 | w.mu.Unlock() |
| 642 | return errors.New("sim: cgroup.kill: the child is not empty") |
| 643 | } |
| 644 | w.killFails[key] = 0 |
| 645 | w.mu.Unlock() |
| 646 | stuck := 0 |
| 647 | for _, p := range procs { |
| 648 | if p.mode == procStubborn || p.mode == procKillFails { |
| 649 | p.exitLater() |
| 650 | stuck++ |
| 651 | continue |
| 652 | } |
| 653 | p.die("cgroup.kill") |
| 654 | } |
| 655 | if stuck > 0 { |
| 656 | return fmt.Errorf("sim: g%d still holds %d process(es)", n, stuck) |
| 657 | } |
| 658 | return nil |
| 659 | } |
| 660 | |
| 661 | func (b *simBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault } |
| 662 | func (b *simBackend) GenReady(registry.Route, Gen, Proc) {} |
| 663 | |
| 664 | // RaiseGenParent refuses without containment, as plan 2's ParentLimit |
| 665 | // does, so such a seed never overlaps. |
| 666 | func (b *simBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error { |
| 667 | if !b.w.contained { |
| 668 | return fmt.Errorf("sim: %w", errNoGenContainment) |
| 669 | } |
| 670 | return nil |
| 671 | } |
| 672 | func (b *simBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error { |
| 673 | return nil |
| 674 | } |
| 675 | func (b *simBackend) GenSwitched(rt registry.Route, g Gen, _ Proc) error { |
| 676 | b.w.logf("switch %s -> gen %d (%s)", rt.Host, g.N, g.Build) |
| 677 | return nil |
| 678 | } |
| 679 | |
| 680 | type simDue struct{} |
| 681 | |
| 682 | func (simDue) SetNextDue(string, int64) error { return nil } |
| 683 | func (simDue) ClearNextDueIf(string, int64) error { return nil } |
| 684 | func (simDue) DueInstances(int64) ([]registry.Route, error) { return nil, nil } |
| 685 | func (simDue) AlwaysOnWorkers() ([]registry.Route, error) { return nil, nil } |
| 686 | func (simDue) SetVersions(string, string, string) error { return nil } |
| 687 | func (simDue) EarliestTimers() (map[string]int64, error) { return nil, nil } |
| 688 | |
| 689 | // checkSwap is I3 on the result: a SwapDone reports the attempt's build and |
| 690 | // staged path. The live link only ever names v1 or v2 and attempts only |
| 691 | // target v3 or v4, so a swap that ran the live link is a visible mismatch. |
| 692 | func (w *simWorld) checkSwap(at Attempt, res SwapResult) { |
| 693 | w.logf("swap %s %s -> %s: %s %s", at.Host, at.Mode, at.Build, res.Outcome, res.Reason) |
| 694 | w.tally(res) |
| 695 | if res.Outcome != SwapDone { |
| 696 | return |
| 697 | } |
| 698 | if res.Gen.Build != at.Build || res.Gen.StagedPath != at.StagedPath { |
| 699 | w.violatef("I3: %s swap of %s to %s reported generation %d running %s from %q", at.Mode, at.Host, at.Build, res.Gen.N, res.Gen.Build, res.Gen.StagedPath) |
| 700 | } |
| 701 | } |
| 702 | |
| 703 | // checkGenerations is I1, polled against the entry's state under a.mu |
| 704 | // until stop closes. |
| 705 | func (w *simWorld) checkGenerations(a *Service, hosts []string, stop <-chan struct{}) { |
| 706 | retired := map[string]map[uint64]bool{} |
| 707 | for { |
| 708 | for _, h := range hosts { |
| 709 | cur, members := a.genStateForTest(h) |
| 710 | if retired[h] == nil { |
| 711 | retired[h] = map[uint64]bool{} |
| 712 | } |
| 713 | attempts := 0 |
| 714 | for n, role := range members { |
| 715 | if role == roleRetiring { |
| 716 | retired[h][n] = true |
| 717 | } else { |
| 718 | attempts++ |
| 719 | } |
| 720 | if cur.N != 0 && n == cur.N { |
| 721 | w.violatef("I1: generation %d of %s is current and also %v", n, h, role) |
| 722 | } |
| 723 | } |
| 724 | if attempts > 1 { |
| 725 | w.violatef("I1: %d attempts in flight for %s", attempts, h) |
| 726 | } |
| 727 | if cur.N != 0 && retired[h][cur.N] { |
| 728 | w.violatef("I1: retired generation %d of %s is current again", cur.N, h) |
| 729 | } |
| 730 | } |
| 731 | select { |
| 732 | case <-stop: |
| 733 | return |
| 734 | case <-time.After(200 * time.Microsecond): |
| 735 | } |
| 736 | } |
| 737 | } |
| 738 | |
| 739 | func onOff(b bool) string { |
| 740 | if b { |
| 741 | return "on" |
| 742 | } |
| 743 | return "off" |
| 744 | } |
| 745 | |
| 746 | func simulateSwapOnce(seed int, staged map[string]simArtifact, sockDir string, sink *simLogSink) ([]string, []string, map[string]int) { |
| 747 | hosts := []string{"one.sim.example.com", "two.sim.example.com"} |
| 748 | w := &simWorld{rng: rand.New(rand.NewSource(int64(seed))), running: map[string]map[*simProc]bool{}, |
| 749 | live: map[string]string{}, staged: map[string]string{}, seen: map[string]bool{}, outcomes: map[string]int{}, |
| 750 | killFails: map[string]int{}, hosts: hosts} |
| 751 | // One seed in five runs with the dormant switch off (today's paths, |
| 752 | // where stop()'s own ErrStillRunning check is the only retention), and |
| 753 | // one in five without generation containment. |
| 754 | w.gensOn, w.contained = w.rng.Intn(5) != 0, w.rng.Intn(5) != 0 |
| 755 | w.outcomes[fmt.Sprintf("seeds with generations %s, containment %s", onOff(w.gensOn), onOff(w.contained))]++ |
| 756 | w.events = append(w.events, fmt.Sprintf("seed %d: generations %s, containment %s", seed, onOff(w.gensOn), onOff(w.contained))) |
| 757 | sink.w.Store(w) |
| 758 | defer sink.w.Store(nil) |
| 759 | for b, art := range staged { |
| 760 | w.staged[b] = art.path |
| 761 | } |
| 762 | for _, h := range hosts { |
| 763 | w.live[h] = "v1" |
| 764 | } |
| 765 | sb := &simBackend{w: w, dir: sockDir, reps: map[string]*simRep{}} |
| 766 | clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| 767 | a := New(Config{Bucket: "sim", Idle: time.Hour}, sb, simDue{}) |
| 768 | a.gens = w.gensOn |
| 769 | a.now = clk.now |
| 770 | a.genHealth = func(ctx context.Context, _ Gen) error { |
| 771 | w.jitter() |
| 772 | if err := ctx.Err(); err != nil { |
| 773 | return err |
| 774 | } |
| 775 | if w.intn(6) == 0 { |
| 776 | return errors.New("sim: unhealthy") |
| 777 | } |
| 778 | return nil |
| 779 | } |
| 780 | revs := map[string]*atomic.Uint64{} |
| 781 | for _, h := range hosts { |
| 782 | revs[h] = new(atomic.Uint64) |
| 783 | } |
| 784 | newest := func(h string) uint64 { return revs[h].Load() } |
| 785 | |
| 786 | stopCheck, checked := make(chan struct{}), make(chan struct{}) |
| 787 | go func() { defer close(checked); w.checkGenerations(a, hosts, stopCheck) }() |
| 788 | |
| 789 | var reqs, swaps sync.WaitGroup |
| 790 | var fenceGen atomic.Uint64 |
| 791 | // The sweeper. The real Service ticks it on its own goroutine, so a |
| 792 | // pending stop or an unproven wake is retried without an event to |
| 793 | // trigger it. Here it ticks every 5ms of wall time (the fake clock does |
| 794 | // not move, so nothing goes idle by it) and is serialized with the |
| 795 | // events' own sweeps: there is one sweeper. It stops before Shutdown, as |
| 796 | // the real one does. |
| 797 | var sweepMu sync.Mutex |
| 798 | sweep := func() { |
| 799 | sweepMu.Lock() |
| 800 | a.sweepOnce() |
| 801 | sweepMu.Unlock() |
| 802 | } |
| 803 | stopSweeper, sweeperDone := make(chan struct{}), make(chan struct{}) |
| 804 | go func() { |
| 805 | defer close(sweeperDone) |
| 806 | tick := time.NewTicker(5 * time.Millisecond) |
| 807 | defer tick.Stop() |
| 808 | for { |
| 809 | select { |
| 810 | case <-stopSweeper: |
| 811 | return |
| 812 | case <-tick.C: |
| 813 | sweep() |
| 814 | } |
| 815 | } |
| 816 | }() |
| 817 | for i := 0; i < 25; i++ { |
| 818 | host := hosts[w.intn(len(hosts))] |
| 819 | rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(sockDir, host+".sock"), |
| 820 | DBPath: filepath.Join(sockDir, host+".db"), Hibernate: true, Status: "ready"} |
| 821 | switch w.intn(12) { |
| 822 | case 0, 1: |
| 823 | reqs.Add(1) |
| 824 | go func() { |
| 825 | defer reqs.Done() |
| 826 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 827 | defer cancel() |
| 828 | if release, err := a.beginRequest(ctx, rt); err == nil { |
| 829 | w.jitter() |
| 830 | release() |
| 831 | } |
| 832 | }() |
| 833 | case 2, 3, 4: |
| 834 | mode := ModeOverlap |
| 835 | if w.intn(3) == 0 { |
| 836 | mode = ModeHold |
| 837 | } |
| 838 | build := []string{"v3", "v4"}[w.intn(2)] |
| 839 | at := Attempt{Host: host, Cause: CauseRelease, Build: build, StagedPath: staged[build].path, Digest: staged[build].digest, |
| 840 | Mode: mode, SchedRev: revs[host].Add(1), Canary: w.intn(2) == 0} |
| 841 | // A scheduler swaps awake instances. Three attempts in four wake |
| 842 | // their host first, so most reach a spawn, a switch or a hold's |
| 843 | // pinned wake; the rest meet whatever state the host is in. |
| 844 | wakeFirst := w.intn(4) != 0 |
| 845 | swaps.Add(1) |
| 846 | go func() { |
| 847 | defer swaps.Done() |
| 848 | if wakeFirst { |
| 849 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 850 | if release, err := a.beginRequest(ctx, rt); err == nil { |
| 851 | release() |
| 852 | } |
| 853 | cancel() |
| 854 | } |
| 855 | w.checkSwap(at, a.Swap(context.Background(), at, newest)) |
| 856 | }() |
| 857 | case 5: |
| 858 | clk.advance(2 * time.Hour) |
| 859 | sweep() |
| 860 | case 6: |
| 861 | sb.loseLease(host) |
| 862 | sweep() |
| 863 | case 7: |
| 864 | a.Bounce(host) |
| 865 | case 8: |
| 866 | reqs.Add(1) |
| 867 | go func() { |
| 868 | defer reqs.Done() |
| 869 | g := fenceGen.Add(1) |
| 870 | ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| 871 | defer cancel() |
| 872 | _, _ = a.FenceAndStop(ctx, host, g) |
| 873 | w.jitter() |
| 874 | a.LiftFence(host, g) |
| 875 | }() |
| 876 | case 9: |
| 877 | w.setLive(host, []string{"v1", "v2"}[w.intn(2)]) |
| 878 | revs[host].Add(1) // the scheduler's newer desired build supersedes in-flight attempts |
| 879 | case 10: |
| 880 | w.crashOne(host) |
| 881 | sweep() |
| 882 | case 11: |
| 883 | // A stop-retry interval passes: the sweep's re-stop cadences |
| 884 | // (pending stops, unproven members) come due, and nothing goes |
| 885 | // idle. |
| 886 | clk.advance(actStopTimeout + time.Second) |
| 887 | sweep() |
| 888 | } |
| 889 | w.jitter() |
| 890 | } |
| 891 | reqs.Wait() |
| 892 | close(stopSweeper) |
| 893 | <-sweeperDone |
| 894 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 895 | a.Shutdown(ctx) |
| 896 | cancel() |
| 897 | w.mu.Lock() |
| 898 | w.shutdown = true |
| 899 | w.mu.Unlock() |
| 900 | w.logf("Shutdown returned") |
| 901 | // "Nothing runs after Shutdown returns" is the generations-on Shutdown |
| 902 | // (spec §7.4: it cancels and joins waking entries, mutant M6's line). |
| 903 | // With the switch off, today's rule holds by design (global constraints: |
| 904 | // Shutdown and wake() byte for byte as today): a wake already in flight |
| 905 | // resolves on its own and the lease TTL bounds it. Flag-off seeds check |
| 906 | // every lease release and acquire, not this clause. |
| 907 | if n := w.runningCount(); n > 0 && w.gensOn { |
| 908 | w.violatef("I2: %d generation process(es) still run after Shutdown returned", n) |
| 909 | } |
| 910 | done := make(chan struct{}) |
| 911 | go func() { swaps.Wait(); close(done) }() |
| 912 | select { |
| 913 | case <-done: |
| 914 | case <-time.After(10 * time.Second): |
| 915 | w.violatef("liveness: a Swap did not return after Shutdown") |
| 916 | } |
| 917 | if n := w.runningCount(); n > 0 && w.gensOn { |
| 918 | w.violatef("I2: %d generation process(es) run once every Swap has returned after Shutdown", n) |
| 919 | } |
| 920 | close(stopCheck) |
| 921 | <-checked |
| 922 | return w.snapshot() |
| 923 | } |
| 924 | |
| 925 | // The model's death is one transition (final review F2): a reaper that |
| 926 | // sees a process Exited() and releases its lease never finds the same |
| 927 | // process still listed as running, so the sweep cannot report a false I2. |
| 928 | // The dying hook runs at the one point a reaper could act during a death; |
| 929 | // a second, concurrent die of the same process returns there too. |
| 930 | func TestSimDeathIsOneTransition(t *testing.T) { |
| 931 | for _, concurrent := range []bool{false, true} { |
| 932 | w := &simWorld{rng: rand.New(rand.NewSource(1)), running: map[string]map[*simProc]bool{}, |
| 933 | live: map[string]string{}, staged: map[string]string{}, seen: map[string]bool{}, outcomes: map[string]int{}, |
| 934 | killFails: map[string]int{}} |
| 935 | p := w.spawn("h.example.com", "v1", 1, true) |
| 936 | fired := false |
| 937 | w.dying = func(q *simProc) { |
| 938 | if fired { |
| 939 | return |
| 940 | } |
| 941 | fired = true |
| 942 | if concurrent { |
| 943 | q.die("concurrent") |
| 944 | } |
| 945 | // A reaper: the process reads as exited, so it gives the lease back. |
| 946 | if q.Exited() { |
| 947 | w.leaseFreed(q.host, "reaper") |
| 948 | } |
| 949 | } |
| 950 | p.die("first") |
| 951 | if !fired { |
| 952 | t.Fatalf("concurrent %v: the dying hook never ran", concurrent) |
| 953 | } |
| 954 | if v, ev, _ := w.snapshot(); len(v) != 0 { |
| 955 | t.Fatalf("concurrent %v: the model reported %v\n%s", concurrent, v, strings.Join(ev, "\n")) |
| 956 | } |
| 957 | if n := w.runningOn("h.example.com"); n != 0 { |
| 958 | t.Fatalf("concurrent %v: %d process(es) still listed after the death", concurrent, n) |
| 959 | } |
| 960 | } |
| 961 | } |
| 962 | |
amadan