| new file mode 100644 |
| index 000000000..3511a6b5e |
| --- /dev/null |
| +++ b/internal/rollout/sim_test.go |
| @@ -0,0 +1,956 @@ |
| +package rollout |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "math/rand" |
| + "os" |
| + "runtime/debug" |
| + "sort" |
| + "strconv" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/activator" |
| + "github.com/carlosframework/platform/internal/appconfig" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// TestSimInvariants is plan 5's invariant simulation (spec §12 |
| +// "Concurrency"). It drives the real Manager, scheduler, appQueue, |
| +// FailureBook, failCohort, decideWake and planCheck through random schedules |
| +// of promotions, agent passes, admissions, attempt outcomes, planned wakes and |
| +// their first-wake observations, hibernation, restarts, config deliveries, |
| +// `carlos promote --retry`, `carlos instances retry` and restaging. Only the |
| +// activator is modelled (simAct, and the wake steps below), along with root's |
| +// install (a promotion writes the rows' installed build) and the scheduler's |
| +// goroutine, whose loop body (tick) runs as a step of its own. Each attempt |
| +// runs the real scheduler.execute and scheduler.done on a goroutine of its |
| +// own, handed off so that exactly one goroutine runs at any moment: a seed is |
| +// one deterministic interleaving, and every invariant check sees a state no |
| +// transition is halfway through. |
| +// |
| +// The invariants, checked against the model's own ground truth (what it |
| +// promoted, what failed, what it retried), never against the book: |
| +// |
| +// - I1: a release-cause failure never leaves an instance that has a good |
| +// build unable to boot one: every wake of such an instance boots |
| +// something, a first wake in overlap mode carries the good build as its |
| +// fallback, and no wake boots, for an instance that has a good build and |
| +// has not served it, a build whose failure on this box stands. Spec |
| +// §6.4's hold-down (a hold build that ran and failed on the instance) is |
| +// the one refusal it allows, and spec §8.5's bootstrap rule exempts an |
| +// instance with no good build. |
| +// - I2: swap_concurrency is never exceeded. |
| +// - I3: a failure record never blocks an instance already serving that |
| +// build. With capacity free, its restart and config attempts onto that |
| +// build run unless their own cohort's canary is still unsettled on |
| +// another host, and an attempt by an instance that serves (or has as its |
| +// good build) the build it failed on never writes or changes a record. |
| +// - I4: a swap never switches to a superseded build: at the switch, an |
| +// adoption's build is the instance's installed build, and no failure of |
| +// it stands against the instance. |
| +// |
| +// Seeds: ROLLOUT_SIM_SEEDS (default 200; the per-revision sweep is 3000). |
| +// ROLLOUT_SIM_SEED=<n> replays one seed. The production code iterates some |
| +// maps, so a replay is best effort; the action log printed with a failure is |
| +// the evidence. Every violation names its invariant ("I2 violated at seed"), |
| +// which is what testdata/rolloutsim/mutate.sh reads. |
| +func TestSimInvariants(t *testing.T) { |
| + first, last := int64(0), int64(200) |
| + if v := os.Getenv("ROLLOUT_SIM_SEEDS"); v != "" { |
| + n, err := strconv.ParseInt(v, 10, 64) |
| + if err != nil || n < 1 { |
| + t.Fatalf("ROLLOUT_SIM_SEEDS=%q", v) |
| + } |
| + last = n |
| + } |
| + if v := os.Getenv("ROLLOUT_SIM_SEED"); v != "" { |
| + n, err := strconv.ParseInt(v, 10, 64) |
| + if err != nil { |
| + t.Fatalf("ROLLOUT_SIM_SEED=%q", v) |
| + } |
| + first, last = n, n+1 |
| + } |
| + opt := t.TempDir() |
| + staged := map[string]bool{} |
| + failed := 0 |
| + for seed := first; seed < last; seed++ { |
| + v := runSimSeed(seed, opt, staged, simSteps) |
| + if v == nil { |
| + continue |
| + } |
| + failed++ |
| + t.Errorf("rolloutsim: %s\n--- last actions ---\n%s", v.line, strings.Join(v.trace, "\n")) |
| + if failed >= 3 { |
| + t.FailNow() |
| + } |
| + } |
| +} |
| + |
| +const simSteps = 400 |
| + |
| +// The activator's first-wake window, as spawnPlanned reads it: B's health |
| +// window ends firstWakeBudget less firstWakeKillReserve after the wake's |
| +// start, and B is exec'd only with firstWakeMinVerdict of it left. |
| +const ( |
| + simHealthWindow = 55*time.Second - 25*time.Second |
| + simMinVerdict = 20 * time.Second |
| +) |
| + |
| +// simViolation is an invariant violation, panicked by violate and recovered |
| +// per seed. Any other panic is a crash of the simulation or the code under |
| +// test, reported apart so the mutation script never counts it as a kill. |
| +type simViolation struct { |
| + inv, msg string |
| +} |
| + |
| +type simCrash struct { |
| + msg string |
| + stack []byte |
| +} |
| + |
| +type simFailure struct { |
| + line string |
| + trace []string |
| +} |
| + |
| +func runSimSeed(seed int64, opt string, staged map[string]bool, steps int) (out *simFailure) { |
| + var s *sim |
| + defer func() { |
| + r := recover() |
| + if r == nil { |
| + return |
| + } |
| + step, trace := -1, []string(nil) |
| + if s != nil { |
| + step, trace = s.step, s.trace |
| + } |
| + switch v := r.(type) { |
| + case simViolation: |
| + out = &simFailure{line: fmt.Sprintf("%s violated at seed %d step %d: %s", v.inv, seed, step, v.msg), trace: trace} |
| + case simCrash: |
| + out = &simFailure{line: fmt.Sprintf("crash at seed %d step %d: %s\n%s", seed, step, v.msg, v.stack), trace: trace} |
| + default: |
| + out = &simFailure{line: fmt.Sprintf("crash at seed %d step %d: %v\n%s", seed, step, r, debug.Stack()), trace: trace} |
| + } |
| + }() |
| + s = newSim(seed, opt, staged) |
| + s.run(steps) |
| + return nil |
| +} |
| + |
| +func violate(inv, format string, args ...any) { |
| + panic(simViolation{inv: inv, msg: fmt.Sprintf(format, args...)}) |
| +} |
| + |
| +// simPhase is an instance's state as the modelled activator holds it. |
| +type simPhase int |
| + |
| +const ( |
| + simAsleep simPhase = iota |
| + simWaking // planned, not yet exec'd: owned, nothing current |
| + simObserving // a first wake inside its 10 s observation: owned, Current hides it |
| + simAwake // serving gen |
| + simStopping // hibernating: owned, nothing current |
| +) |
| + |
| +type simHost struct { |
| + name string |
| + ch int |
| + phase simPhase |
| + gen activator.Gen // the generation serving (awake) or observed (observing) |
| + // good is the model's good build: the last build that settled here. It |
| + // is ground truth, kept apart from the row the code writes. |
| + good string |
| + served map[string]bool |
| + // A wake in flight: its plan, the route the activator holds for it, and |
| + // the FailedUnder the model expects for its build. |
| + plan activator.WakePlan |
| + planRoute registry.Route |
| + wakeStart time.Time |
| + stamp time.Time |
| + failLater bool // observing: the build exits inside its observation |
| +} |
| + |
| +type simBuild struct { |
| + bad bool // fails on every instance that has not served it |
| + flaky bool // fails half the time on an instance that has not served it |
| + holdDemand bool // the release demands hold |
| + misstaged bool // the staged file is not the canonical artifact until root restages |
| +} |
| + |
| +// simSwap is one attempt inside the modelled Swap, parked until a complete |
| +// step lets it finish. |
| +type simSwap struct { |
| + a activator.Attempt |
| + newest func(string) uint64 |
| + release chan struct{} |
| + stamp time.Time // the FailedUnder the model expects |
| + // Set by finishSwap for the I3 record check: whether the instance |
| + // serves the attempt's build, and the book's record for it beforehand. |
| + serves bool |
| + before string |
| +} |
| + |
| +type simEvent struct { |
| + host string |
| + swap *simSwap // the attempt reached Swap and is parked |
| + crash any |
| + stack []byte |
| +} |
| + |
| +type sim struct { |
| + r *rand.Rand |
| + seed int64 |
| + step int |
| + now time.Time |
| + trace []string |
| + |
| + opt string |
| + staged map[string]bool |
| + m *Manager |
| + sched *scheduler |
| + book *FailureBook |
| + rows *fakeRoutes |
| + |
| + hosts []*simHost |
| + byName map[string]*simHost |
| + channels [2]string |
| + installed [2]string |
| + pointers map[string]release.ChannelPointer // channel -> its signed pointer |
| + builds map[string]*simBuild |
| + nb, nreq int |
| + hold bool // the app's swap policy is hold |
| + concurrency int |
| + genN uint64 |
| + |
| + // Ground truth. A failure of build b stands while no pointer naming b |
| + // has been signed with a PromotedAt newer than b's newest failure stamp |
| + // (the FailedUnder its attempt or wake read, else the failure's time): |
| + // spec rev 14 §8.4. exempt is `carlos instances retry` per host until |
| + // that host fails the build again; heldDown is a hold build that ran and |
| + // failed on the host, until a host retry. |
| + lastStamp map[string]time.Time |
| + lastSigned map[string]time.Time |
| + failedEver map[string]bool |
| + exempt map[string]map[string]bool |
| + heldDown map[string]map[string]bool |
| + |
| + swaps map[string]*simSwap |
| + events chan simEvent |
| + draining bool |
| +} |
| + |
| +func newSim(seed int64, opt string, staged map[string]bool) *sim { |
| + s := &sim{r: rand.New(rand.NewSource(seed)), seed: seed, now: t0, opt: opt, staged: staged, |
| + book: &FailureBook{}, rows: &fakeRoutes{rows: map[string]registry.Route{}}, |
| + byName: map[string]*simHost{}, channels: [2]string{"", "beta"}, |
| + pointers: map[string]release.ChannelPointer{}, builds: map[string]*simBuild{}, |
| + lastStamp: map[string]time.Time{}, lastSigned: map[string]time.Time{}, failedEver: map[string]bool{}, |
| + exempt: map[string]map[string]bool{}, heldDown: map[string]map[string]bool{}, |
| + swaps: map[string]*simSwap{}, events: make(chan simEvent)} |
| + s.hold = s.r.Intn(4) == 0 |
| + s.concurrency = 1 + s.r.Intn(3) |
| + s.addBuild("b0", &simBuild{}) |
| + s.installed = [2]string{"b0", "b0"} |
| + for _, ch := range s.channels { |
| + s.sign(ch, "b0") |
| + } |
| + n := 3 + s.r.Intn(4) |
| + for i := range n { |
| + h := &simHost{name: "h" + strconv.Itoa(i), ch: i % 2, served: map[string]bool{}} |
| + row := registry.Route{Host: h.name, Kind: "instance", App: "blog", Account: "bab", Status: "ready", |
| + Hibernate: true, Version: "b0", Channel: s.channels[h.ch]} |
| + switch { |
| + case s.r.Intn(2) == 0: |
| + // Awake: Bootstrap records the build it serves as good (spec §8.5). |
| + h.phase, h.gen = simAwake, s.newGen(s.identity("b0")) |
| + h.good, h.served["b0"] = "b0", true |
| + case s.r.Intn(2) == 0: |
| + id := s.identity("b0") |
| + row.GoodVersion, row.GoodDigest, row.GoodDemand = id.Version, id.Digest, id.Demand |
| + h.good, h.served["b0"] = "b0", true |
| + default: |
| + // Asleep with no good build: the upgrade's bootstrap case. |
| + } |
| + s.rows.put(row) |
| + s.hosts = append(s.hosts, h) |
| + s.byName[h.name] = h |
| + } |
| + s.m = NewManager(context.Background(), Deps{Act: simAct{s}, Routes: s.rows, Book: s.book, OptDir: opt, Box: "box-1", |
| + Demand: s.demand, Policy: s.policy, Pointer: s.pointer, Now: func() time.Time { return s.now }, |
| + Logf: func(string, ...any) {}}) |
| + // The app's scheduler, registered by hand so its goroutine never starts: |
| + // tick is its loop body, run as a step. The queue is built as |
| + // Manager.scheduler builds it. |
| + key := appKey{"bab", "blog"} |
| + book := s.book |
| + s.sched = &scheduler{m: s.m, key: key, kick: make(chan struct{}, 1), |
| + q: newAppQueue(key.acct, key.app, func(build, host string, served bool) bool { |
| + return book.Blocks(key.acct, key.app, build, host, served) |
| + })} |
| + s.m.apps[key] = s.sched |
| + rs, _ := s.rows.List() |
| + s.m.Bootstrap(rs) |
| + s.pass() |
| + return s |
| +} |
| + |
| +func (s *sim) logf(format string, args ...any) { |
| + if len(s.trace) >= 60 { |
| + s.trace = s.trace[1:] |
| + } |
| + s.trace = append(s.trace, fmt.Sprintf("%4d %s ", s.step, s.now.Sub(t0))+fmt.Sprintf(format, args...)) |
| +} |
| + |
| +func (s *sim) addBuild(b string, sb *simBuild) { |
| + s.builds[b] = sb |
| + if !s.staged[b] { |
| + if err := release.StageArtifact(s.opt, "bab", b, "blog", []byte(b)); err != nil { |
| + panic(simCrash{msg: "stage " + b + ": " + err.Error()}) |
| + } |
| + s.staged[b] = true |
| + } |
| +} |
| + |
| +func (s *sim) identity(b string) release.StagedIdentity { |
| + id := release.StagedIdentity{Version: b, StagedPath: release.VersionPath(s.opt, "bab", b, "blog"), Digest: "d-" + b} |
| + if s.builds[b].holdDemand { |
| + id.Demand = release.DemandHold |
| + } |
| + return id |
| +} |
| + |
| +func (s *sim) newGen(id release.StagedIdentity) activator.Gen { |
| + s.genN++ |
| + return activator.Gen{N: s.genN, Build: id.Version, StagedPath: id.StagedPath, Digest: id.Digest, Demand: id.Demand} |
| +} |
| + |
| +func (s *sim) sign(channel, b string) { |
| + s.pointers[channel] = release.ChannelPointer{Schema: 1, Account: "bab", App: "blog", Channel: channel, Version: b, PromotedAt: s.now} |
| + s.lastSigned[b] = s.now |
| +} |
| + |
| +// demand is release.CanonicalDemand's contract, as the harness has it. |
| +func (s *sim) demand(_ context.Context, _, _, ver, _ string) (string, string, error) { |
| + b := s.builds[ver] |
| + if b == nil || b.misstaged { |
| + return "", "", fmt.Errorf("%w: the staged executable's digest differs from the canonical manifest", release.ErrStagedDigestMismatch) |
| + } |
| + if b.holdDemand { |
| + return release.DemandHold, "d-" + ver, nil |
| + } |
| + return "", "d-" + ver, nil |
| +} |
| + |
| +func (s *sim) policy(context.Context, string, string) (*appconfig.Manifest, error) { |
| + swap := appconfig.SwapOverlap |
| + if s.hold { |
| + swap = appconfig.SwapHold |
| + } |
| + return &appconfig.Manifest{Schema: 1, App: "blog", Swap: swap, SwapConcurrency: s.concurrency}, nil |
| +} |
| + |
| +func (s *sim) pointer(_ context.Context, _, _, channel string) (*release.ChannelPointer, error) { |
| + p, ok := s.pointers[channel] |
| + if !ok { |
| + return nil, nil |
| + } |
| + return &p, nil |
| +} |
| + |
| +// stampFor is the FailedUnder the model expects for build on host, read when |
| +// its attempt or wake starts: the PromotedAt of the host's channel pointer |
| +// when it names build, else zero (the failure's own time stands in). |
| +func (s *sim) stampFor(h *simHost, build string) time.Time { |
| + if p, ok := s.pointers[s.channels[h.ch]]; ok && p.Version == build { |
| + return p.PromotedAt |
| + } |
| + return time.Time{} |
| +} |
| + |
| +func (s *sim) expectHold(build string) bool { |
| + return s.hold || s.builds[build].holdDemand |
| +} |
| + |
| +// failure is the ground truth of a build that ran and failed on host. |
| +func (s *sim) failure(build, host string, stamp time.Time, heldDown bool) { |
| + if stamp.IsZero() { |
| + stamp = s.now |
| + } |
| + s.failedEver[build] = true |
| + if stamp.After(s.lastStamp[build]) { |
| + s.lastStamp[build] = stamp |
| + } |
| + delete(s.exempt[host], build) |
| + if heldDown { |
| + if s.heldDown[host] == nil { |
| + s.heldDown[host] = map[string]bool{} |
| + } |
| + s.heldDown[host][build] = true |
| + } |
| + s.logf("ground truth: %s failed on %s (stamp %s, held down %v)", build, host, stamp.Sub(t0), heldDown) |
| +} |
| + |
| +func (s *sim) stands(build string) bool { |
| + return s.failedEver[build] && !s.lastSigned[build].After(s.lastStamp[build]) |
| +} |
| + |
| +func (s *sim) blockedFor(host, build string) bool { |
| + return s.stands(build) && !s.exempt[host][build] |
| +} |
| + |
| +// snap is the book's record for build, as text, for the I3 record check. |
| +func (s *sim) snap(build string) string { |
| + r, ok := s.book.Lookup("bab", "blog", build) |
| + return fmt.Sprintf("%v %+v", ok, r) |
| +} |
| + |
| +// simAct is the modelled activator. Only one goroutine runs at a time (the |
| +// handoffs in launch and complete), so it reads the model without a lock. |
| +type simAct struct{ s *sim } |
| + |
| +func (f simAct) Current(host string) (activator.Gen, bool) { |
| + if h := f.s.byName[host]; h != nil && h.phase == simAwake { |
| + return h.gen, true |
| + } |
| + return activator.Gen{}, false |
| +} |
| + |
| +func (f simAct) Owns(host string) bool { |
| + h := f.s.byName[host] |
| + return h != nil && h.phase != simAsleep |
| +} |
| + |
| +// Swap parks the attempt until a complete step releases it. |
| +func (f simAct) Swap(_ context.Context, a activator.Attempt, newest func(string) uint64) activator.SwapResult { |
| + sw := &simSwap{a: a, newest: newest, release: make(chan struct{})} |
| + f.s.events <- simEvent{host: a.Host, swap: sw} |
| + <-sw.release |
| + return f.s.finishSwap(sw) |
| +} |
| + |
| +func (f simAct) ReplaceWorker(context.Context, activator.Attempt, *release.StagedIdentity, func(string) uint64) activator.SwapResult { |
| + panic(simCrash{msg: "the simulation runs no workers"}) |
| +} |
| + |
| +func (f simAct) PublishHeldServing(string) {} |
| + |
| +// finishSwap is the modelled Swap's end (plan 3), run on the attempt's |
| +// goroutine while the simulation waits: the re-check against newest, then |
| +// the build's health, then the switch. It is one transition: nothing else |
| +// runs until done() has returned. |
| +func (s *sim) finishSwap(sw *simSwap) activator.SwapResult { |
| + a := sw.a |
| + h := s.byName[a.Host] |
| + sw.serves = h.good == a.Build || (h.phase == simAwake && h.gen.Build == a.Build) |
| + sw.before = s.snap(a.Build) |
| + res := activator.SwapResult{Mode: a.Mode} |
| + if s.draining || sw.newest(a.Host) != a.SchedRev || s.r.Intn(30) == 0 { |
| + // Superseded, or a re-check that refused (a lost lease, a stop). |
| + res.Outcome, res.Reason = activator.SwapAbandoned, "superseded" |
| + s.logf("swap %s %s %s: abandoned", a.Host, a.Cause, a.Build) |
| + return res |
| + } |
| + fresh := !h.served[a.Build] |
| + b := s.builds[a.Build] |
| + fails := false |
| + switch { |
| + case fresh: |
| + fails = b.bad || (b.flaky && s.r.Intn(2) == 0) |
| + case h.phase == simAwake && a.Build == h.gen.Build: |
| + fails = s.r.Intn(8) == 0 // a restart or config attempt that fails |
| + } |
| + if fails { |
| + res.Reason = "health check failed" |
| + if a.Mode == activator.ModeHold && s.r.Intn(4) != 0 { |
| + // The old generation was stopped first: nothing serves. |
| + res.Outcome = activator.SwapDown |
| + h.phase, h.gen = simAsleep, activator.Gen{} |
| + } else { |
| + res.Outcome = activator.SwapFailed |
| + } |
| + s.logf("swap %s %s %s (%s): %s", a.Host, a.Cause, a.Build, a.Mode, res.Outcome) |
| + if fresh { |
| + s.failure(a.Build, a.Host, sw.stamp, res.Outcome == activator.SwapDown) |
| + } |
| + return res |
| + } |
| + if a.Cause == activator.CauseRelease || a.Cause == activator.CauseRetry { |
| + if want := s.installed[h.ch]; a.Build != want { |
| + violate("I4", "%s switched to %s (cause %s) while its installed build is %s", a.Host, a.Build, a.Cause, want) |
| + } |
| + if a.Build != h.good && !h.served[a.Build] && s.blockedFor(a.Host, a.Build) { |
| + violate("I4", "%s switched to %s (cause %s) after its failure on this box was recorded", a.Host, a.Build, a.Cause) |
| + } |
| + } |
| + g := s.newGen(release.StagedIdentity{Version: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand}) |
| + h.phase, h.gen = simAwake, g |
| + h.good, h.served[a.Build] = a.Build, true |
| + res.Outcome, res.Gen = activator.SwapDone, g |
| + s.logf("swap %s %s %s (%s): done", a.Host, a.Cause, a.Build, a.Mode) |
| + return res |
| +} |
| + |
| +// await is the handoff: the one running attempt goroutine has parked in Swap |
| +// or returned. |
| +func (s *sim) await() simEvent { |
| + select { |
| + case ev := <-s.events: |
| + if ev.crash != nil { |
| + if v, ok := ev.crash.(simViolation); ok { |
| + panic(v) |
| + } |
| + if c, ok := ev.crash.(simCrash); ok { |
| + panic(c) |
| + } |
| + panic(simCrash{msg: fmt.Sprint(ev.crash), stack: ev.stack}) |
| + } |
| + return ev |
| + case <-time.After(30 * time.Second): |
| + panic(simCrash{msg: "an attempt neither reached Swap nor returned within 30s"}) |
| + } |
| +} |
| + |
| +// launch runs the real scheduler.execute for d up to its Swap (or to its |
| +// return, for a staging refusal). |
| +func (s *sim) launch(d dispatch) { |
| + stamp := s.stampFor(s.byName[d.ID.Host], d.ID.Build) |
| + go func() { |
| + defer func() { |
| + ev := simEvent{host: d.ID.Host} |
| + if r := recover(); r != nil { |
| + ev.crash, ev.stack = r, debug.Stack() |
| + } |
| + s.events <- ev |
| + }() |
| + s.sched.execute(context.Background(), d) |
| + }() |
| + ev := s.await() |
| + if ev.swap == nil { |
| + s.logf("launch %s %s %s: returned before Swap", d.ID.Host, d.ID.Cause, d.ID.Build) |
| + return |
| + } |
| + ev.swap.stamp = stamp |
| + s.swaps[d.ID.Host] = ev.swap |
| + s.logf("launch %s %s %s rev %d canary %v class %d", d.ID.Host, d.ID.Cause, d.ID.Build, d.SchedRev, d.Canary, d.Class) |
| +} |
| + |
| +func (s *sim) run(steps int) { |
| + for s.step = 0; s.step < steps; s.step++ { |
| + s.now = s.now.Add(time.Duration(1+s.r.Intn(3)) * time.Second) |
| + switch k := s.r.Intn(26); { |
| + case k < 2: |
| + s.promote() |
| + case k < 4: |
| + s.pass() |
| + case k < 10: |
| + s.tick() |
| + case k < 14: |
| + s.complete() |
| + case k < 16: |
| + s.wakeBegin() |
| + case k < 18: |
| + s.wakeExec() |
| + case k < 19: |
| + s.observeEnd() |
| + case k < 20: |
| + s.sleep() |
| + case k < 21: |
| + s.stopEnd() |
| + case k < 22: |
| + s.restart() |
| + case k < 23: |
| + s.config() |
| + case k < 24: |
| + s.pointerRetry() |
| + case k < 25: |
| + s.hostRetry() |
| + default: |
| + s.restage() |
| + } |
| + s.checkI2() |
| + } |
| + s.draining = true |
| + for _, host := range sortedKeys(s.swaps) { |
| + sw := s.swaps[host] |
| + delete(s.swaps, host) |
| + close(sw.release) |
| + s.await() |
| + } |
| +} |
| + |
| +func sortedKeys[V any](m map[string]V) []string { |
| + out := make([]string, 0, len(m)) |
| + for k := range m { |
| + out = append(out, k) |
| + } |
| + sort.Strings(out) |
| + return out |
| +} |
| + |
| +func (s *sim) pick(ok func(h *simHost) bool) *simHost { |
| + var c []*simHost |
| + for _, h := range s.hosts { |
| + if ok(h) { |
| + c = append(c, h) |
| + } |
| + } |
| + if len(c) == 0 { |
| + return nil |
| + } |
| + return c[s.r.Intn(len(c))] |
| +} |
| + |
| +// pass is the agent pass (Manager.Pass) over every row. |
| +func (s *sim) pass() { |
| + rs, _ := s.rows.List() |
| + s.m.Pass(rs) |
| +} |
| + |
| +// promote is a promotion and root's install onto one channel's hosts, then |
| +// the agent pass that sees it: a fresh build, or the other channel's build |
| +// promoted here (which signs a newer pointer for it, spec rev 14 §8.4). |
| +func (s *sim) promote() { |
| + ch := s.r.Intn(2) |
| + var b string |
| + if other := s.installed[1-ch]; s.r.Intn(4) == 0 && other != s.installed[ch] { |
| + b = other |
| + } else { |
| + s.nb++ |
| + b = "b" + strconv.Itoa(s.nb) |
| + s.addBuild(b, &simBuild{bad: s.r.Intn(4) == 0, flaky: s.r.Intn(5) == 0, holdDemand: s.r.Intn(6) == 0, misstaged: s.r.Intn(10) == 0}) |
| + } |
| + s.installed[ch] = b |
| + s.sign(s.channels[ch], b) |
| + for _, h := range s.hosts { |
| + if h.ch == ch { |
| + row := s.rows.get(h.name) |
| + row.Version = b |
| + s.rows.put(row) |
| + } |
| + } |
| + s.logf("promote %s onto channel %q %+v", b, s.channels[ch], *s.builds[b]) |
| + s.pass() |
| +} |
| + |
| +// tick is scheduler.loop's body: the view, configSlept, admit, and the |
| +// attempts it hands out run up to their Swap. |
| +func (s *sim) tick() { |
| + view, pendingConfig := s.sched.view() |
| + for _, host := range sortedKeys(pendingConfig) { |
| + if serving, seen := view[host]; seen && serving == "" { |
| + s.m.configSlept(host, pendingConfig[host]) |
| + } |
| + } |
| + s.sched.mu.Lock() |
| + ds := s.sched.q.admit(view, s.now) |
| + s.sched.mu.Unlock() |
| + // Liveness is judged on the admission itself, before any attempt runs: |
| + // one that returns at once (a staging refusal) frees its slot after |
| + // admit stopped at the limit, and the loop's next pass admits the rest. |
| + s.checkI2() |
| + s.checkI3Liveness(view) |
| + for _, d := range ds { |
| + s.launch(d) |
| + } |
| + s.checkI2() |
| +} |
| + |
| +func (s *sim) checkI2() { |
| + s.sched.mu.Lock() |
| + running := s.sched.q.running |
| + s.sched.mu.Unlock() |
| + if len(s.swaps) > s.concurrency || running > s.concurrency { |
| + violate("I2", "%d attempts in Swap (%d running in the queue) with swap_concurrency %d", len(s.swaps), running, s.concurrency) |
| + } |
| +} |
| + |
| +// checkI3Liveness runs after an admission that left capacity free: every |
| +// awake instance's restart or config attempt onto the build it serves was |
| +// admitted, unless it waits out its own deferral or its own cohort's |
| +// unsettled canary on another host. A failure record (of the build it serves |
| +// or of the one it wants) never holds it. |
| +func (s *sim) checkI3Liveness(view map[string]string) { |
| + q := s.sched.q |
| + s.sched.mu.Lock() |
| + defer s.sched.mu.Unlock() |
| + if q.running >= s.concurrency { |
| + return |
| + } |
| + for _, h := range s.hosts { |
| + if serving, seen := view[h.name]; !seen || serving == "" { |
| + continue |
| + } |
| + slot := q.hosts[h.name] |
| + if slot == nil || slot.running != nil { |
| + continue |
| + } |
| + d, ok := q.pending(slot) |
| + if !ok || d.ID.Build != slot.serving { |
| + continue |
| + } |
| + if s.now.Before(slot.notBefore) || (d.ID.Cause == activator.CauseConfig && s.now.Before(slot.configAfter)) { |
| + continue |
| + } |
| + if c := q.cohorts[cohortKey(slot.serving, classRecover)]; c != nil && c.canary != "" && c.canary != h.name && !c.settled { |
| + continue |
| + } |
| + violate("I3", "%s's %s attempt onto %s, the build it serves, was held back with capacity free (cohort %s)", h.name, d.ID.Cause, d.ID.Build, d.Cohort) |
| + } |
| +} |
| + |
| +// complete lets one parked attempt finish: the modelled Swap's end, then the |
| +// real done(). |
| +func (s *sim) complete() { |
| + hosts := sortedKeys(s.swaps) |
| + if len(hosts) == 0 { |
| + return |
| + } |
| + host := hosts[s.r.Intn(len(hosts))] |
| + sw := s.swaps[host] |
| + delete(s.swaps, host) |
| + close(sw.release) |
| + if ev := s.await(); ev.swap != nil { |
| + panic(simCrash{msg: "a second attempt reached Swap while " + host + "'s finished"}) |
| + } |
| + if sw.serves && s.snap(sw.a.Build) != sw.before { |
| + violate("I3", "%s's %s attempt onto %s, a build it serves, changed its failure record: %s -> %s", host, sw.a.Cause, sw.a.Build, sw.before, s.snap(sw.a.Build)) |
| + } |
| +} |
| + |
| +// heldDownOn reports whether the model let host be held down on build: a |
| +// hold build that ran and failed there, with no host retry since. |
| +func (s *sim) heldDownOn(host, build string) bool { |
| + return s.heldDown[host][build] |
| +} |
| + |
| +// wakeBegin is a wake's planning (Manager.Plan, as the activator calls it |
| +// before the lease). |
| +func (s *sim) wakeBegin() { |
| + h := s.pick(func(h *simHost) bool { return h.phase == simAsleep }) |
| + if h == nil { |
| + return |
| + } |
| + row := s.rows.get(h.name) |
| + h.phase = simWaking |
| + plan := s.m.Plan(context.Background(), row) |
| + if plan.Refuse != "" || plan.Boot.StagedPath == "" { |
| + h.phase = simAsleep |
| + s.logf("wake %s: refused at planning: %s", h.name, plan.Refuse) |
| + s.checkWakeRefused(h, row.Version, "its plan refused: "+plan.Refuse) |
| + return |
| + } |
| + h.plan, h.planRoute, h.wakeStart, h.stamp = plan, row, s.now, s.stampFor(h, plan.Boot.Version) |
| + s.logf("wake %s planned: boot %s first %v fallback %v mode %s", h.name, plan.Boot.Version, plan.FirstWake, plan.Fallback != nil, plan.Mode) |
| +} |
| + |
| +func (s *sim) checkWakeRefused(h *simHost, installed, why string) { |
| + if h.good != "" && !s.heldDownOn(h.name, installed) { |
| + violate("I1", "%s has good build %s but its wake boots nothing: %s", h.name, h.good, why) |
| + } |
| +} |
| + |
| +// checkBoot is I1 at the exec of build on h. |
| +func (s *sim) checkBoot(h *simHost, build string) { |
| + if h.good != "" && build != h.good && !h.served[build] && s.blockedFor(h.name, build) { |
| + violate("I1", "%s (good build %s) boots %s, whose failure on this box stands", h.name, h.good, build) |
| + } |
| +} |
| + |
| +// wakeExec is the activator's checkPlan and exec (spec §7.4, §8.2): the |
| +// plan's Check, one replan if it refuses, the verdict floor, and the build's |
| +// outcome reported through the real OnGen. |
| +func (s *sim) wakeExec() { |
| + h := s.pick(func(h *simHost) bool { return h.phase == simWaking }) |
| + if h == nil { |
| + return |
| + } |
| + rt, plan := h.planRoute, h.plan |
| + if s.r.Intn(20) == 0 { |
| + // The lease was refused (a previous incarnation still holds it): the |
| + // wake ends before any exec and reports no outcome, so only a Pass |
| + // that finds the activator no longer owns the host lets go of the |
| + // build its plan carried (Manager.planned). Not a build failure, so |
| + // no I1 check. |
| + h.phase = simAsleep |
| + s.logf("wake %s: refused at its lease, no outcome reported", h.name) |
| + return |
| + } |
| + if plan.Check != nil && plan.Check(plan.Boot.Version) != nil { |
| + next := s.m.Plan(context.Background(), rt) |
| + if next.Refuse != "" || next.Boot.StagedPath == "" || (next.Check != nil && next.Check(next.Boot.Version) != nil) { |
| + h.phase = simAsleep |
| + s.logf("wake %s: abandoned at exec (plan stale, replan %q)", h.name, next.Refuse) |
| + s.checkWakeRefused(h, rt.Version, "the plan went stale and the replan refused") |
| + return |
| + } |
| + plan, h.plan, h.stamp = next, next, s.stampFor(h, next.Boot.Version) |
| + s.logf("wake %s replanned: boot %s first %v", h.name, plan.Boot.Version, plan.FirstWake) |
| + } |
| + canFall := plan.FirstWake && plan.Fallback != nil && plan.Mode == activator.ModeOverlap |
| + if left := h.wakeStart.Add(simHealthWindow).Sub(s.now); canFall && left < simMinVerdict { |
| + // The verdict floor (spawnPlanned): B's health window ends 30 s |
| + // after the wake's start (55 s budget less the 25 s kill reserve), |
| + // and with less than 20 s of it left at the exec, the good build |
| + // boots untried and nothing is recorded against B. |
| + good := *plan.Fallback |
| + s.checkBoot(h, good.Version) |
| + h.phase, h.gen = simAwake, s.newGen(good) |
| + h.served[good.Version] = true |
| + s.logf("wake %s: %s untried, booted the good build %s", h.name, plan.Boot.Version, good.Version) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now}) |
| + return |
| + } |
| + boot := plan.Boot |
| + s.checkBoot(h, boot.Version) |
| + if h.good != "" && plan.FirstWake && !s.expectHold(boot.Version) && (plan.Fallback == nil || plan.Fallback.Version != h.good) { |
| + violate("I1", "%s (good build %s) first-wakes %s in overlap mode without its good build behind it", h.name, h.good, boot.Version) |
| + } |
| + if !plan.FirstWake { |
| + h.phase, h.gen = simAwake, s.newGen(boot) |
| + h.good, h.served[boot.Version] = boot.Version, true |
| + s.logf("wake %s: booted %s", h.name, boot.Version) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, At: s.now}) |
| + return |
| + } |
| + b := s.builds[boot.Version] |
| + fails := b.bad || (b.flaky && s.r.Intn(2) == 0) |
| + if fails && s.r.Intn(2) == 0 { |
| + // It passed its health check and exits inside its observation. |
| + fails, h.failLater = false, true |
| + } else { |
| + h.failLater = false |
| + } |
| + switch { |
| + case !fails: |
| + h.phase, h.gen = simObserving, s.newGen(boot) |
| + s.logf("wake %s: first wake of %s healthy, observing", h.name, boot.Version) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now}) |
| + case canFall: |
| + s.failure(boot.Version, h.name, h.stamp, false) |
| + h.phase, h.gen = simAwake, s.newGen(*plan.Fallback) |
| + s.logf("wake %s: first wake of %s failed, fell back to %s", h.name, boot.Version, plan.Fallback.Version) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, Gen: h.gen, Failed: boot, Mode: plan.Mode, |
| + FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder}) |
| + default: |
| + hold := plan.Mode == activator.ModeHold |
| + s.failure(boot.Version, h.name, h.stamp, hold) |
| + h.phase, h.gen = simAsleep, activator.Gen{} |
| + s.logf("wake %s: first wake of %s failed, nothing fell back (mode %s)", h.name, boot.Version, plan.Mode) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: boot, Mode: plan.Mode, |
| + FirstWake: true, Reason: "health check failed", At: s.now, FailedUnder: plan.FailedUnder}) |
| + if h.good != "" && !hold { |
| + violate("I1", "%s (good build %s) failed a first wake of %s in overlap mode and nothing fell back", h.name, h.good, boot.Version) |
| + } |
| + } |
| +} |
| + |
| +// observeEnd ends a first wake's observation: it settles, or the build exits. |
| +func (s *sim) observeEnd() { |
| + h := s.pick(func(h *simHost) bool { return h.phase == simObserving }) |
| + if h == nil { |
| + return |
| + } |
| + rt, plan := h.planRoute, h.plan |
| + if h.failLater { |
| + s.failure(plan.Boot.Version, h.name, h.stamp, plan.Mode == activator.ModeHold) |
| + h.phase, h.gen = simAsleep, activator.Gen{} |
| + s.logf("wake %s: %s exited inside its observation", h.name, plan.Boot.Version) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true, |
| + Reason: "exited within 10s of becoming healthy", At: s.now, FailedUnder: plan.FailedUnder}) |
| + return |
| + } |
| + h.phase = simAwake |
| + h.good, h.served[h.gen.Build] = h.gen.Build, true |
| + s.logf("wake %s: %s settled", h.name, h.gen.Build) |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: h.gen, Mode: plan.Mode, FirstWake: true, At: s.now}) |
| +} |
| + |
| +// sleep hibernates an instance with no attempt in Swap: at once, or through a |
| +// stopping phase the view leaves out. An observation it ends is cancelled. |
| +func (s *sim) sleep() { |
| + h := s.pick(func(h *simHost) bool { |
| + _, inSwap := s.swaps[h.name] |
| + return !inSwap && (h.phase == simAwake || h.phase == simObserving) |
| + }) |
| + if h == nil { |
| + return |
| + } |
| + observing := h.phase == simObserving |
| + g := h.gen |
| + h.phase, h.gen = simAsleep, activator.Gen{} |
| + if s.r.Intn(2) == 0 { |
| + h.phase = simStopping |
| + } |
| + s.logf("sleep %s (stopping phase %v, observation cancelled %v)", h.name, h.phase == simStopping, observing) |
| + if observing { |
| + s.m.OnGen(activator.GenEvent{Kind: activator.GenCancelled, Route: h.planRoute, Gen: g, Mode: h.plan.Mode, FirstWake: true, At: s.now}) |
| + } |
| +} |
| + |
| +func (s *sim) stopEnd() { |
| + if h := s.pick(func(h *simHost) bool { return h.phase == simStopping }); h != nil { |
| + h.phase = simAsleep |
| + s.logf("stopped %s", h.name) |
| + } |
| +} |
| + |
| +func (s *sim) restart() { |
| + h := s.hosts[s.r.Intn(len(s.hosts))] |
| + s.nreq++ |
| + id := "r" + strconv.Itoa(s.nreq) |
| + ok := s.m.Restart(h.name, id) |
| + s.logf("restart %s %s: %v", h.name, id, ok) |
| +} |
| + |
| +func (s *sim) config() { |
| + h := s.hosts[s.r.Intn(len(s.hosts))] |
| + s.nreq++ |
| + rev := "c" + strconv.Itoa(s.nreq) |
| + s.m.ConfigDelivered(h.name, rev) |
| + s.logf("config %s %s", h.name, rev) |
| +} |
| + |
| +// pointerRetry is `carlos promote --retry`: the channel's head is signed again |
| +// with a newer PromotedAt. The next due agent pass clears the record. |
| +func (s *sim) pointerRetry() { |
| + ch := s.r.Intn(2) |
| + b := s.installed[ch] |
| + if !s.stands(b) { |
| + return |
| + } |
| + s.sign(s.channels[ch], b) |
| + s.logf("promote --retry %s on channel %q", b, s.channels[ch]) |
| +} |
| + |
| +// hostRetry is `carlos instances retry <host>` for a host whose installed |
| +// build's failure stands. |
| +func (s *sim) hostRetry() { |
| + h := s.hosts[s.r.Intn(len(s.hosts))] |
| + b := s.installed[h.ch] |
| + if !s.stands(b) { |
| + return |
| + } |
| + ok := s.m.RetryHost(h.name) |
| + if s.exempt[h.name] == nil { |
| + s.exempt[h.name] = map[string]bool{} |
| + } |
| + s.exempt[h.name][b] = true |
| + delete(s.heldDown[h.name], b) |
| + s.logf("instances retry %s onto %s: %v", h.name, b, ok) |
| +} |
| + |
| +// restage is root restaging a build whose staged file was wrong. |
| +func (s *sim) restage() { |
| + for _, b := range sortedKeys(s.builds) { |
| + if s.builds[b].misstaged { |
| + s.builds[b].misstaged = false |
| + s.logf("restaged %s", b) |
| + return |
| + } |
| + } |
| +} |
amadan