| new file mode 100644 |
| index 000000000..e3a89a27b |
| --- /dev/null |
| +++ b/internal/activator/swap_sim_test.go |
| @@ -0,0 +1,602 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "errors" |
| + "fmt" |
| + "log" |
| + "math/rand" |
| + "os" |
| + "path/filepath" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/cgroup" |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// TestSwapInvariants drives random schedules of requests, overlap and hold |
| +// swaps, hibernation, lease loss, restarts, maintenance fences, root |
| +// repoints, supersession, crashes and a final Shutdown against the real |
| +// Service and a simulated backend. It checks I1, I2 and I3 (plan 3 task 13; |
| +// spec §12). |
| +// |
| +// Seeds: SWAP_SIM_SEEDS (default 100). SWAP_SIM_SEED=<n> replays one seed. |
| +// Goroutine interleaving is not deterministic, so a replay is best effort. |
| +// The event log printed with a failure is the evidence. |
| +// |
| +// The mutation checks (testdata/swapsim/mutate.sh) run this sweep and the |
| +// TestSwapWitness* tests against each `// swapsim:Mn` mutant. |
| +func TestSwapInvariants(t *testing.T) { |
| + first, last := 0, 100 |
| + if v := os.Getenv("SWAP_SIM_SEEDS"); v != "" { |
| + n, err := strconv.Atoi(v) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + last = n |
| + } |
| + if v := os.Getenv("SWAP_SIM_SEED"); v != "" { |
| + n, err := strconv.Atoi(v) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + first, last = n, n+1 |
| + } |
| + oldSettle, oldJoin := swapSettle, opJoinBound |
| + swapSettle, opJoinBound = 2*time.Millisecond, 3*time.Second |
| + t.Cleanup(func() { swapSettle, opJoinBound = oldSettle, oldJoin }) |
| + // The Service's own log lines go into the seed's event log, so a failure |
| + // prints them in order with the model's events, and a passing sweep |
| + // prints nothing. |
| + sink := &simLogSink{} |
| + oldOut, oldFlags := log.Writer(), log.Flags() |
| + log.SetOutput(sink) |
| + log.SetFlags(0) |
| + t.Cleanup(func() { log.SetOutput(oldOut); log.SetFlags(oldFlags) }) |
| + staged := stageSimBuilds(t) |
| + sockDir := t.TempDir() |
| + failed := 0 |
| + tally := map[string]int{} |
| + t.Cleanup(func() { |
| + t.Logf("swap outcomes over seeds %d..%d (mode/outcome, as Swap reported them): %v", first, last-1, tally) |
| + }) |
| + for seed := first; seed < last; seed++ { |
| + violations, events, outcomes := simulateSwapOnce(seed, staged, sockDir, sink) |
| + for k, n := range outcomes { |
| + tally[k] += n |
| + } |
| + if len(violations) == 0 { |
| + continue |
| + } |
| + failed++ |
| + if len(events) > 200 { |
| + events = events[len(events)-200:] |
| + } |
| + t.Errorf("invariant violated at seed %d: %s\n--- last events ---\n%s", seed, strings.Join(violations, "; "), strings.Join(events, "\n")) |
| + if failed >= 3 { |
| + t.FailNow() |
| + } |
| + } |
| +} |
| + |
| +type simArtifact struct{ path, digest string } |
| + |
| +func stageSimBuilds(t *testing.T) map[string]simArtifact { |
| + t.Helper() |
| + out := map[string]simArtifact{} |
| + for _, b := range []string{"v3", "v4"} { |
| + dir := filepath.Join(t.TempDir(), "versions", b) |
| + if err := os.MkdirAll(dir, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + p := filepath.Join(dir, "app") |
| + body := []byte("build " + b) |
| + if err := os.WriteFile(p, body, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + sum := sha256.Sum256(body) |
| + out[b] = simArtifact{p, hex.EncodeToString(sum[:])} |
| + } |
| + return out |
| +} |
| + |
| +// simLogSink routes the log package into the current seed's world. A |
| +// goroutine a previous seed left behind may log into the next seed's |
| +// events; the lines say what they are. |
| +type simLogSink struct{ w atomic.Pointer[simWorld] } |
| + |
| +func (s *simLogSink) Write(p []byte) (int, error) { |
| + if w := s.w.Load(); w != nil { |
| + w.logf("log: %s", strings.TrimRight(string(p), "\n")) |
| + } |
| + return len(p), nil |
| +} |
| + |
| +// simWorld is the model: which simulated processes run on each host, the |
| +// live link, the staged builds, and the violations seen. Lock order: a |
| +// caller may hold a.mu when it takes w.mu (the backend is called under the |
| +// entry lock), never the reverse. |
| +type simWorld struct { |
| + mu sync.Mutex |
| + rng *rand.Rand |
| + events []string |
| + running map[string]map[*simProc]bool |
| + live map[string]string |
| + staged map[string]string |
| + violations []string |
| + seen map[string]bool |
| + nextID int |
| + // shutdown is set once Service.Shutdown has returned: nothing may spawn |
| + // after it (I2). |
| + shutdown bool |
| + // outcomes counts Swap results by the mode they ran in and their |
| + // outcome, so a sweep shows it reached switches and holds. |
| + outcomes map[string]int |
| +} |
| + |
| +func (w *simWorld) logf(format string, args ...any) { |
| + w.mu.Lock() |
| + w.events = append(w.events, fmt.Sprintf(format, args...)) |
| + w.mu.Unlock() |
| +} |
| + |
| +func (w *simWorld) violatef(format string, args ...any) { |
| + msg := fmt.Sprintf(format, args...) |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + if w.seen[msg] { |
| + return |
| + } |
| + w.seen[msg] = true |
| + w.violations = append(w.violations, msg) |
| + w.events = append(w.events, "VIOLATION "+msg) |
| +} |
| + |
| +func (w *simWorld) intn(n int) int { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + return w.rng.Intn(n) |
| +} |
| + |
| +func (w *simWorld) jitter() { |
| + if d := w.intn(4); d > 0 { |
| + time.Sleep(time.Duration(d) * 250 * time.Microsecond) |
| + } |
| +} |
| + |
| +func (w *simWorld) spawn(host, build string, n uint64) *simProc { |
| + w.mu.Lock() |
| + w.nextID++ |
| + p := &simProc{w: w, id: w.nextID, host: host, build: build, n: n} |
| + if w.running[host] == nil { |
| + w.running[host] = map[*simProc]bool{} |
| + } |
| + w.running[host][p] = true |
| + w.events = append(w.events, fmt.Sprintf("spawn p%d %s build=%s gen=%d", p.id, host, build, n)) |
| + after := w.shutdown |
| + w.mu.Unlock() |
| + if after { |
| + w.violatef("I2: p%d (%s build %s) spawned after Shutdown returned", p.id, host, build) |
| + } |
| + return p |
| +} |
| + |
| +func (w *simWorld) runningOn(host string) int { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + return len(w.running[host]) |
| +} |
| + |
| +func (w *simWorld) runningCount() int { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + n := 0 |
| + for _, ps := range w.running { |
| + n += len(ps) |
| + } |
| + return n |
| +} |
| + |
| +func (w *simWorld) leaseFreed(host, by string) { |
| + if n := w.runningOn(host); n > 0 { |
| + w.violatef("I2: lease of %s released by %s while %d generation(s) run", host, by, n) |
| + } |
| + w.logf("lease %s released by %s", host, by) |
| +} |
| + |
| +func (w *simWorld) setLive(host, build string) { |
| + w.mu.Lock() |
| + w.live[host] = build |
| + w.events = append(w.events, "repoint "+host+" -> "+build) |
| + w.mu.Unlock() |
| +} |
| + |
| +func (w *simWorld) crashOne(host string) { |
| + w.mu.Lock() |
| + var victims []*simProc |
| + for p := range w.running[host] { |
| + victims = append(victims, p) |
| + } |
| + var p *simProc |
| + if len(victims) > 0 { |
| + p = victims[w.rng.Intn(len(victims))] |
| + } |
| + w.mu.Unlock() |
| + if p != nil { |
| + p.die("crash") |
| + } |
| +} |
| + |
| +func (w *simWorld) snapshot() ([]string, []string, map[string]int) { |
| + w.mu.Lock() |
| + defer w.mu.Unlock() |
| + outcomes := make(map[string]int, len(w.outcomes)) |
| + for k, n := range w.outcomes { |
| + outcomes[k] = n |
| + } |
| + return append([]string(nil), w.violations...), append([]string(nil), w.events...), outcomes |
| +} |
| + |
| +// simProc is one simulated generation process. Stop always ends it: a |
| +// termination the sim never leaves unproven, so every retained lease the |
| +// sim sees is a retention the code chose. |
| +type simProc struct { |
| + w *simWorld |
| + id int |
| + host string |
| + build string |
| + n uint64 |
| + mu sync.Mutex |
| + dead bool |
| +} |
| + |
| +func (p *simProc) Stop(context.Context) error { p.w.jitter(); p.die("stop"); return nil } |
| + |
| +func (p *simProc) Exited() bool { |
| + p.mu.Lock() |
| + defer p.mu.Unlock() |
| + return p.dead |
| +} |
| + |
| +func (p *simProc) die(why string) { |
| + p.mu.Lock() |
| + was := p.dead |
| + p.dead = true |
| + p.mu.Unlock() |
| + if was { |
| + return |
| + } |
| + p.w.mu.Lock() |
| + delete(p.w.running[p.host], p) |
| + p.w.events = append(p.w.events, fmt.Sprintf("exit p%d (%s)", p.id, why)) |
| + p.w.mu.Unlock() |
| +} |
| + |
| +type simRep struct { |
| + w *simWorld |
| + host string |
| + mu sync.Mutex |
| + err error |
| +} |
| + |
| +func (r *simRep) Err() error { |
| + r.mu.Lock() |
| + defer r.mu.Unlock() |
| + return r.err |
| +} |
| + |
| +func (r *simRep) Close(context.Context) error { |
| + r.w.jitter() |
| + r.w.leaseFreed(r.host, "rep.Close") |
| + return nil |
| +} |
| + |
| +type simLease struct { |
| + b *simBackend |
| + host string |
| +} |
| + |
| +func (l *simLease) Replicate(context.Context, string, time.Duration) (Rep, error) { |
| + r := &simRep{w: l.b.w, host: l.host} |
| + l.b.mu.Lock() |
| + l.b.reps[l.host] = r |
| + l.b.mu.Unlock() |
| + return r, nil |
| +} |
| + |
| +func (l *simLease) Release(context.Context) error { |
| + l.b.w.leaseFreed(l.host, "lease.Release") |
| + return nil |
| +} |
| + |
| +// simBackend is Backend plus SwapBackend. It is not a GenBackend, so a |
| +// wake of the live link goes through Start and gets canonicalGen (a fresh |
| +// number on the canonical socket), as every pre-generations test double |
| +// does. It has no genKiller, hostKiller or sidecarKiller: a generation's |
| +// leader exit is its proof (genChildEmpty, sidecarTerminated). |
| +type simBackend struct { |
| + w *simWorld |
| + dir string |
| + mu sync.Mutex |
| + reps map[string]*simRep |
| +} |
| + |
| +var ( |
| + _ Backend = (*simBackend)(nil) |
| + _ SwapBackend = (*simBackend)(nil) |
| + _ DueStore = simDue{} |
| +) |
| + |
| +func (b *simBackend) loseLease(host string) { |
| + b.mu.Lock() |
| + r := b.reps[host] |
| + b.mu.Unlock() |
| + if r != nil { |
| + r.mu.Lock() |
| + r.err = errors.New("sim: lease lost") |
| + r.mu.Unlock() |
| + } |
| +} |
| + |
| +func (b *simBackend) Acquire(_ context.Context, rt registry.Route) (Lease, error) { |
| + b.w.jitter() |
| + if n := b.w.runningOn(rt.Host); n > 0 { |
| + b.w.violatef("I2: lease of %s acquired while %d generation(s) of the previous holder run", rt.Host, n) |
| + } |
| + return &simLease{b: b, host: rt.Host}, nil |
| +} |
| +func (b *simBackend) Restore(context.Context, registry.Route) (bool, error) { |
| + b.w.jitter() |
| + return false, nil |
| +} |
| +func (b *simBackend) Seed(context.Context, registry.Route) error { return nil } |
| +func (b *simBackend) Start(_ context.Context, rt registry.Route) (Proc, string, error) { |
| + b.w.jitter() |
| + b.w.mu.Lock() |
| + build := b.w.live[rt.Host] |
| + b.w.mu.Unlock() |
| + return b.w.spawn(rt.Host, build, 0), build, nil |
| +} |
| +func (b *simBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil } |
| +func (b *simBackend) HasSidecar(registry.Route) bool { return false } |
| +func (b *simBackend) StartSidecar(context.Context, registry.Route) (Proc, error) { |
| + return nil, errors.New("sim: no sidecars") |
| +} |
| +func (b *simBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) { |
| + return nil, "", errors.New("sim: no workers") |
| +} |
| + |
| +// StartStaged checks I3 at the exec: the staged path a generation runs is |
| +// the one its build was staged at. Its sockets live in the run's own |
| +// directory, because a hold's pinned wake links the canonical path to the |
| +// generation's socket for real (swapCanonicalLink). |
| +func (b *simBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) { |
| + b.w.mu.Lock() |
| + want := b.w.staged[g.Build] |
| + b.w.mu.Unlock() |
| + if g.StagedPath != want { |
| + 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) |
| + } |
| + b.w.jitter() |
| + if err := ctx.Err(); err != nil { |
| + return g, nil, err |
| + } |
| + if b.w.intn(20) == 0 { |
| + return g, nil, errors.New("sim: exec failed") |
| + } |
| + g.Socket = filepath.Join(b.dir, fmt.Sprintf("%s.%d.sock", rt.Host, g.N)) |
| + g.Control = g.Socket + ".control" |
| + return g, b.w.spawn(rt.Host, g.Build, g.N), nil |
| +} |
| +func (b *simBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault } |
| +func (b *simBackend) GenReady(registry.Route, Gen, Proc) {} |
| +func (b *simBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *simBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *simBackend) GenSwitched(rt registry.Route, g Gen, _ Proc) error { |
| + b.w.logf("switch %s -> gen %d (%s)", rt.Host, g.N, g.Build) |
| + return nil |
| +} |
| + |
| +type simDue struct{} |
| + |
| +func (simDue) SetNextDue(string, int64) error { return nil } |
| +func (simDue) ClearNextDueIf(string, int64) error { return nil } |
| +func (simDue) DueInstances(int64) ([]registry.Route, error) { return nil, nil } |
| +func (simDue) AlwaysOnWorkers() ([]registry.Route, error) { return nil, nil } |
| +func (simDue) SetVersions(string, string, string) error { return nil } |
| +func (simDue) EarliestTimers() (map[string]int64, error) { return nil, nil } |
| + |
| +// checkSwap is I3 on the result: a SwapDone reports the attempt's build and |
| +// staged path. The live link only ever names v1 or v2 and attempts only |
| +// target v3 or v4, so a swap that ran the live link is a visible mismatch. |
| +func (w *simWorld) checkSwap(at Attempt, res SwapResult) { |
| + w.logf("swap %s %s -> %s: %s %s", at.Host, at.Mode, at.Build, res.Outcome, res.Reason) |
| + w.mu.Lock() |
| + w.outcomes[string(res.Mode)+"/"+string(res.Outcome)]++ |
| + w.mu.Unlock() |
| + if res.Outcome != SwapDone { |
| + return |
| + } |
| + if res.Gen.Build != at.Build || res.Gen.StagedPath != at.StagedPath { |
| + 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) |
| + } |
| +} |
| + |
| +// checkGenerations is I1, polled against the entry's state under a.mu |
| +// until stop closes. |
| +func (w *simWorld) checkGenerations(a *Service, hosts []string, stop <-chan struct{}) { |
| + retired := map[string]map[uint64]bool{} |
| + for { |
| + for _, h := range hosts { |
| + cur, members := a.genStateForTest(h) |
| + if retired[h] == nil { |
| + retired[h] = map[uint64]bool{} |
| + } |
| + attempts := 0 |
| + for n, role := range members { |
| + if role == roleRetiring { |
| + retired[h][n] = true |
| + } else { |
| + attempts++ |
| + } |
| + if cur.N != 0 && n == cur.N { |
| + w.violatef("I1: generation %d of %s is current and also %v", n, h, role) |
| + } |
| + } |
| + if attempts > 1 { |
| + w.violatef("I1: %d attempts in flight for %s", attempts, h) |
| + } |
| + if cur.N != 0 && retired[h][cur.N] { |
| + w.violatef("I1: retired generation %d of %s is current again", cur.N, h) |
| + } |
| + } |
| + select { |
| + case <-stop: |
| + return |
| + case <-time.After(200 * time.Microsecond): |
| + } |
| + } |
| +} |
| + |
| +func simulateSwapOnce(seed int, staged map[string]simArtifact, sockDir string, sink *simLogSink) ([]string, []string, map[string]int) { |
| + w := &simWorld{rng: rand.New(rand.NewSource(int64(seed))), running: map[string]map[*simProc]bool{}, |
| + live: map[string]string{}, staged: map[string]string{}, seen: map[string]bool{}, outcomes: map[string]int{}} |
| + sink.w.Store(w) |
| + defer sink.w.Store(nil) |
| + for b, art := range staged { |
| + w.staged[b] = art.path |
| + } |
| + hosts := []string{"one.sim.example.com", "two.sim.example.com"} |
| + for _, h := range hosts { |
| + w.live[h] = "v1" |
| + } |
| + sb := &simBackend{w: w, dir: sockDir, reps: map[string]*simRep{}} |
| + clk := &fakeClock{t: time.Unix(1_700_000_000, 0)} |
| + a := New(Config{Bucket: "sim", Idle: time.Hour}, sb, simDue{}) |
| + a.gens = true |
| + a.now = clk.now |
| + a.genHealth = func(ctx context.Context, _ Gen) error { |
| + w.jitter() |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + if w.intn(6) == 0 { |
| + return errors.New("sim: unhealthy") |
| + } |
| + return nil |
| + } |
| + revs := map[string]*atomic.Uint64{} |
| + for _, h := range hosts { |
| + revs[h] = new(atomic.Uint64) |
| + } |
| + newest := func(h string) uint64 { return revs[h].Load() } |
| + |
| + stopCheck, checked := make(chan struct{}), make(chan struct{}) |
| + go func() { defer close(checked); w.checkGenerations(a, hosts, stopCheck) }() |
| + |
| + var reqs, swaps sync.WaitGroup |
| + var fenceGen atomic.Uint64 |
| + for i := 0; i < 25; i++ { |
| + host := hosts[w.intn(len(hosts))] |
| + rt := registry.Route{Host: host, Kind: "instance", Socket: filepath.Join(sockDir, host+".sock"), |
| + DBPath: filepath.Join(sockDir, host+".db"), Hibernate: true, Status: "ready"} |
| + switch w.intn(11) { |
| + case 0, 1: |
| + reqs.Add(1) |
| + go func() { |
| + defer reqs.Done() |
| + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + defer cancel() |
| + if release, err := a.beginRequest(ctx, rt); err == nil { |
| + w.jitter() |
| + release() |
| + } |
| + }() |
| + case 2, 3, 4: |
| + mode := ModeOverlap |
| + if w.intn(3) == 0 { |
| + mode = ModeHold |
| + } |
| + build := []string{"v3", "v4"}[w.intn(2)] |
| + at := Attempt{Host: host, Cause: CauseRelease, Build: build, StagedPath: staged[build].path, Digest: staged[build].digest, |
| + Mode: mode, SchedRev: revs[host].Add(1), Canary: w.intn(2) == 0} |
| + // A scheduler swaps awake instances. Three attempts in four wake |
| + // their host first, so most reach a spawn, a switch or a hold's |
| + // pinned wake; the rest meet whatever state the host is in. |
| + wakeFirst := w.intn(4) != 0 |
| + swaps.Add(1) |
| + go func() { |
| + defer swaps.Done() |
| + if wakeFirst { |
| + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + if release, err := a.beginRequest(ctx, rt); err == nil { |
| + release() |
| + } |
| + cancel() |
| + } |
| + w.checkSwap(at, a.Swap(context.Background(), at, newest)) |
| + }() |
| + case 5: |
| + clk.advance(2 * time.Hour) |
| + a.sweepOnce() |
| + case 6: |
| + sb.loseLease(host) |
| + a.sweepOnce() |
| + case 7: |
| + a.Bounce(host) |
| + case 8: |
| + reqs.Add(1) |
| + go func() { |
| + defer reqs.Done() |
| + g := fenceGen.Add(1) |
| + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) |
| + defer cancel() |
| + _, _ = a.FenceAndStop(ctx, host, g) |
| + w.jitter() |
| + a.LiftFence(host, g) |
| + }() |
| + case 9: |
| + w.setLive(host, []string{"v1", "v2"}[w.intn(2)]) |
| + revs[host].Add(1) // the scheduler's newer desired build supersedes in-flight attempts |
| + case 10: |
| + w.crashOne(host) |
| + a.sweepOnce() |
| + } |
| + w.jitter() |
| + } |
| + reqs.Wait() |
| + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| + a.Shutdown(ctx) |
| + cancel() |
| + w.mu.Lock() |
| + w.shutdown = true |
| + w.mu.Unlock() |
| + w.logf("Shutdown returned") |
| + if n := w.runningCount(); n > 0 { |
| + w.violatef("I2: %d generation(s) still run after Shutdown returned", n) |
| + } |
| + done := make(chan struct{}) |
| + go func() { swaps.Wait(); close(done) }() |
| + select { |
| + case <-done: |
| + case <-time.After(10 * time.Second): |
| + w.violatef("liveness: a Swap did not return after Shutdown") |
| + } |
| + if n := w.runningCount(); n > 0 { |
| + w.violatef("I2: %d generation(s) run once every Swap has returned after Shutdown", n) |
| + } |
| + close(stopCheck) |
| + <-checked |
| + return w.snapshot() |
| +} |