amadan
provision: RecordView invariant simulation; five mutation checks fail as they must (#336)
TestRecordViewSimulation: bumping, bypassing and failed-bump writers, incremental Refresh, Rebuild, sibling watch polls and a moves-pass consumer under a fake w.pass, on DirStore, with invariants (a)-(e) checked after every operation. Codex 1, 2 and 4 are scripted subtests under both ListDirs shapes. -seeds (default 200) and -seed; skipped under -short. 200 seeds: 10.3 s here (fsync-bound, 8 seeds at once). Unmutated 3000-seed sweep, nice 19: pass, 2m36s. Mutation checks (tree copy under $TMPDIR, 3000-seed sweep, nice 19): 1. bump before write (DirStore Put/PutFrom/Delete, release/store.go): FAILS, seed 1, (a). Caught via the in-flight PutFrom write. 2. applied recorded for a failed scope (Refresh `if r.ok`, recordview.go): FAILS, scripted codex4 (e); sweep seed 7, (d) and (e). The weaker form that records applied but keeps the pending set survives 3000 seeds: it is equivalent, since pending re-reads the scope on every Refresh. 3. index read moved after the walk in rebuildLocked: FAILS, scripted codex2 (a); sweep seed 4, (a). 4. Rebuild walks and publishes without mu: FAILS, scripted codex2 (c, revert); sweep seed 18, (c). 5. consumer acts on the handed-over snapshot (simConsumeHandedOver, the simulated moves pass): FAILS, scripted codex1 (c, consumed epoch went back); sweep seed 1, (c).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 file changed,
+1362
−0
internal/provision/recordview_sim_test.go+1362 −0
diff --git a/internal/provision/recordview_sim_test.go b/internal/provision/recordview_sim_test.go| new file mode 100644 |
| index 000000000..3fedd0553 |
| --- /dev/null |
| +++ b/internal/provision/recordview_sim_test.go |
| @@ -0,0 +1,1362 @@ |
| +package provision |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "flag" |
| + "fmt" |
| + "io" |
| + "log" |
| + "math/rand/v2" |
| + "os" |
| + "path/filepath" |
| + "reflect" |
| + "slices" |
| + "sort" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/domains" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// The RecordView invariant simulation (S3 change-index design §8). Random |
| +// interleavings of writers (bumping, bypassing, failing their bump), |
| +// incremental refreshes, backstop rebuilds and a moves-pass consumer run on a |
| +// DirStore, and five invariants are checked after every step: |
| +// |
| +// (a) after a Refresh that returned nil, every write whose bump succeeded |
| +// and that completed before the Refresh's index poll is in the snapshot |
| +// (or that key answers a read failure, which no consumer acts on); |
| +// (b) after a clean Rebuild (no read failed, no write landed during it), |
| +// the record walk over the snapshot equals the record walk over the |
| +// bucket, ListAllRecordsStrict included; |
| +// (c) published epochs only increase, the consumer never consumes a lower |
| +// epoch than it already consumed, and no key ever reverts to an older |
| +// write; |
| +// (d) a snapshot with Complete == false never answers ListAllRecordsStrict |
| +// with a nil error when its gap lies in what that reader reads (an |
| +// account, an instances listing, a record body); a gap anywhere else |
| +// (a sidecar listing, a manifest) is still visible as a read error to |
| +// that family's reader; and a Complete snapshot answers every reader |
| +// without error; |
| +// (e) a scope whose read failed is read again by the next incremental |
| +// Refresh, whether or not the index moved. |
| +// |
| +// How the interleavings are made. The view reads through simStore, which |
| +// carries the current operation in its ctx and calls the simulation before |
| +// every read: a read may fail by the operation's plan, and a trigger planned |
| +// for "before the n-th read of scope S" fires there, writing to S, polling |
| +// the shared watch as a sibling consumer would, or starting a concurrent |
| +// Refresh. The concurrent Refresh runs nested, right there, if RecordView.mu |
| +// is free (TryLock), which is exactly when a real concurrent caller would |
| +// get in; otherwise it would block on mu, so it is queued and runs after the |
| +// operation returns. Triggers fire in the reading goroutine, before the read, |
| +// and only write to the scope being read, so where a write lands relative to |
| +// that scope's reads does not depend on goroutine timing. |
| +// |
| +// A write that bumps and is still in flight (the one interleaving a store |
| +// call cannot expose: between a write and its bump) is a DirStore.PutFrom |
| +// whose body reader blocks on its first Read; refreshes run while it waits. |
| +// A write whose bump fails is a real index-enabled DirStore whose bump the |
| +// filesystem refuses: control/ is read-only for the length of the write. |
| +// Whether a write bumped is never assumed: the index is read before and |
| +// after each write (writes are serialized), so a run as root, where the |
| +// read-only directory does not stop the bump, only moves writes from the |
| +// failed class to the bumped one. |
| +// |
| +// The three interleavings §8 names are scripted subtests that run before the |
| +// sweep, under both ListDirs shapes: codex 2 (the backstop has listed an app |
| +// when a record appears and a concurrent Refresh starts), codex 4 (an app's |
| +// re-read fails after the index moved, and the next index read is a 304) and |
| +// codex 1 (a pass captures a snapshot and stalls while a later pass consumes |
| +// a newer one). A failing seed reruns alone with -seed N. |
| + |
| +var ( |
| + simSeedFlag = flag.Uint64("seed", 0, "TestRecordViewSimulation: run only this seed (0 runs the sweep)") |
| + simSeedsFlag = flag.Int("seeds", 200, "TestRecordViewSimulation: how many seeds the sweep runs") |
| +) |
| + |
| +// simConsumeHandedOver is the consumer's mutation switch. The moves pass |
| +// refreshes and takes the current snapshot inside w.pass and never acts on a |
| +// snapshot handed over from the Reconcile leg (§4.3, codex 1). true makes the |
| +// simulated consumer act on the handed-over one; a mutation check flips it. |
| +var simConsumeHandedOver = false |
| + |
| +var simEpoch = time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| + |
| +var ( |
| + simApps = []string{"shop", "blog", "docs", "wiki", "store", "fresh"} |
| + simSlugs = []string{"one", "two", "three"} |
| + simSidecars = []string{"bot", "cron"} |
| +) |
| + |
| +var errSimRead = errors.New("sim: read refused") |
| + |
| +// simParallel is how many seeds run at once. |
| +const simParallel = 8 |
| + |
| +// simTemplate is the fixture's bucket, read once and copied into a fresh |
| +// root per seed (copying files is far cheaper than writing them through a |
| +// store that fsyncs and bumps). |
| +type simTemplate struct { |
| + files map[string][]byte |
| + acct [3]string |
| +} |
| + |
| +func newSimTemplate(t *testing.T) *simTemplate { |
| + t.Helper() |
| + f := newViewFixture(t, false, true) |
| + keys, err := f.bypass.List(ctx, "") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + tmpl := &simTemplate{files: map[string][]byte{}, acct: f.acct} |
| + for _, k := range keys { |
| + body, _, err := f.bypass.Get(ctx, k) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + tmpl.files[k] = body |
| + } |
| + if _, ok := tmpl.files[release.ChangeIndexKey]; !ok { |
| + t.Fatal("fixture: no change index") |
| + } |
| + return tmpl |
| +} |
| + |
| +// simOp is one view operation (a Refresh or a Rebuild) as the store sees it. |
| +type simOp struct { |
| + kind string |
| + nested bool |
| + fail map[string]bool // keys and prefixes whose every read fails |
| + triggers []*simTrigger |
| + |
| + // Filled in as it runs, under sim.mu. |
| + polled bool |
| + pollSeq int // writes completed when the op first read the index |
| + walked bool |
| + counts map[string]int |
| + read map[string]bool // scopes it read |
| + failed map[string]bool // scopes with a read that failed |
| + failedAny bool |
| + wrote bool // a trigger wrote during it |
| + nestedRan bool // a concurrent Refresh ran inside it |
| + err error |
| +} |
| + |
| +func newSimOp(kind string) *simOp { |
| + return &simOp{kind: kind, fail: map[string]bool{}, counts: map[string]int{}, read: map[string]bool{}, failed: map[string]bool{}} |
| +} |
| + |
| +type simTrigger struct { |
| + scope string |
| + n int |
| + what string |
| + act func(op *simOp) |
| + fired bool |
| +} |
| + |
| +type simOpKey struct{} |
| + |
| +// simWrite is one planned write: decided by the seeded generator up front, |
| +// so a trigger firing in a reader goroutine never touches the rng. |
| +type simWrite struct { |
| + w *simWriter |
| + key string |
| + del bool |
| + soft bool // an account manifest written soft-deleted |
| +} |
| + |
| +type simWriter struct { |
| + name string |
| + store *release.DirStore |
| + failBump bool |
| +} |
| + |
| +type simVersion struct { |
| + seq int |
| + body []byte // nil: absent |
| +} |
| + |
| +type simHandover struct { |
| + sn *ViewSnapshot |
| + from int // step it was captured at |
| +} |
| + |
| +// simStats counts what a sweep exercised, so a simulation that silently |
| +// stopped exercising something fails instead of passing. |
| +type simStats struct { |
| + steps, writes, bumped, bypassed, failedBumps, paused int |
| + refreshes, rebuilds, triggers, nested, deferred, sibling int |
| + readFailures, checksB, checksE, consumed, incomplete int |
| +} |
| + |
| +func (a *simStats) add(b simStats) { |
| + a.steps += b.steps |
| + a.writes += b.writes |
| + a.bumped += b.bumped |
| + a.bypassed += b.bypassed |
| + a.failedBumps += b.failedBumps |
| + a.paused += b.paused |
| + a.refreshes += b.refreshes |
| + a.rebuilds += b.rebuilds |
| + a.triggers += b.triggers |
| + a.nested += b.nested |
| + a.deferred += b.deferred |
| + a.sibling += b.sibling |
| + a.readFailures += b.readFailures |
| + a.checksB += b.checksB |
| + a.checksE += b.checksE |
| + a.consumed += b.consumed |
| + a.incomplete += b.incomplete |
| +} |
| + |
| +type sim struct { |
| + t *testing.T |
| + rng *rand.Rand |
| + bucket bool |
| + acct [3]string |
| + acctID map[string]int64 |
| + |
| + root string |
| + live *simStore // what the view reads |
| + clean *viewCountingStore // the bucket as a reader sees it, for (b) |
| + raw *release.DirStore // plain reads for bookkeeping |
| + f *viewFixture // for readWalk |
| + watch *release.ChangeWatch |
| + v *RecordView |
| + bumpW *simWriter |
| + bypW *simWriter |
| + failW *simWriter |
| + keys []string // every family key the simulation tracks |
| + step int |
| + writes bool // ops may plan writes (false while a write is in flight) |
| + paused bool |
| + |
| + writeMu sync.Mutex // serializes writes, so a bump is attributable |
| + |
| + mu sync.Mutex // guards the fields below |
| + seq int // writes started |
| + done int // writes completed (writes are serialized: done is the last seq) |
| + hist map[string][]simVersion |
| + bumped map[string][]int // key -> seqs of writes whose bump took |
| + dirty map[string]bool // scopes written since the last operation was planned |
| + deferred int |
| + errs []string |
| + trace []string |
| + stats simStats |
| + |
| + checkMu sync.Mutex |
| + lastEp uint64 |
| + lastVer map[string]int |
| + carry map[string]bool // (e): scopes whose last read failed |
| + pass sync.Mutex // the fake w.pass |
| + consumed uint64 |
| + hand []simHandover |
| +} |
| + |
| +func newSim(t *testing.T, tmpl *simTemplate, seed uint64, bucket bool) *sim { |
| + t.Helper() |
| + root := t.TempDir() |
| + for k, b := range tmpl.files { |
| + p := filepath.Join(root, filepath.FromSlash(k)) |
| + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(p, b, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + s := &sim{ |
| + t: t, rng: rand.New(rand.NewPCG(seed, 0x336)), bucket: bucket, acct: tmpl.acct, root: root, |
| + acctID: map[string]int64{}, writes: true, |
| + hist: map[string][]simVersion{}, bumped: map[string][]int{}, dirty: map[string]bool{}, lastVer: map[string]int{}, carry: map[string]bool{}, |
| + } |
| + for i, a := range tmpl.acct { |
| + s.acctID[a] = int64(301 + i) |
| + } |
| + s.live = &simStore{DirStore: release.NewDirStore(root), bucket: bucket, sim: s} |
| + s.clean = &viewCountingStore{DirStore: release.NewDirStore(root), bucket: bucket} |
| + s.raw = release.NewDirStore(root) |
| + s.f = &viewFixture{live: s.clean, bypass: s.raw, acct: tmpl.acct} |
| + s.watch = release.NewChangeWatch(s.live) |
| + s.v = NewRecordView(s.live, s.watch) |
| + bw, fw := release.NewDirStore(root), release.NewDirStore(root) |
| + bw.EnableChangeIndex() |
| + fw.EnableChangeIndex() |
| + s.bumpW = &simWriter{name: "bumping", store: bw} |
| + s.failW = &simWriter{name: "failing-bump", store: fw, failBump: true} |
| + s.bypW = &simWriter{name: "bypassing", store: release.NewDirStore(root)} |
| + |
| + tracked := map[string]bool{} |
| + for _, a := range s.acct { |
| + tracked[accountManifestKey(a)] = true |
| + for _, b := range simApps { |
| + for _, slug := range simSlugs { |
| + tracked[RecordKey(a, b, slug+"."+b+".example.com")] = true |
| + } |
| + for _, n := range simSidecars { |
| + tracked[SidecarRecordKey(a, b, n)] = true |
| + } |
| + tracked[ManifestKey(a, b)] = true |
| + tracked[SidecarManifestKey(a, b)] = true |
| + tracked[release.AppTombstoneKey(a, b)] = true |
| + } |
| + } |
| + for k := range tmpl.files { |
| + if _, ok := (&ViewSnapshot{}).lookup(k); ok { |
| + tracked[k] = true |
| + } |
| + } |
| + for k := range tracked { |
| + s.keys = append(s.keys, k) |
| + var body []byte |
| + if b, ok := tmpl.files[k]; ok { |
| + body = b |
| + } |
| + s.hist[k] = []simVersion{{seq: 0, body: body}} |
| + } |
| + sort.Strings(s.keys) |
| + return s |
| +} |
| + |
| +func (s *sim) fail(format string, args ...any) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.errs = append(s.errs, fmt.Sprintf(format, args...)) |
| +} |
| + |
| +func (s *sim) tracef(format string, args ...any) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.trace = append(s.trace, fmt.Sprintf("[%d] ", s.step)+fmt.Sprintf(format, args...)) |
| +} |
| + |
| +// stop fails the test with the invariant breaches and the trace so far. |
| +func (s *sim) stop() { |
| + s.t.Helper() |
| + s.mu.Lock() |
| + errs, trace := slices.Clone(s.errs), slices.Clone(s.trace) |
| + s.mu.Unlock() |
| + if len(errs) == 0 { |
| + return |
| + } |
| + if len(trace) > 80 { |
| + trace = trace[len(trace)-80:] |
| + } |
| + s.t.Fatalf("bucket=%v step %d:\n %s\ntrace (last %d):\n %s", s.bucket, s.step, |
| + strings.Join(errs, "\n "), len(trace), strings.Join(trace, "\n ")) |
| +} |
| + |
| +// simScope is the change scope a read of target (a key or a listed prefix) |
| +// belongs to, or "" for the top-level listings and the index. |
| +func simScope(target string) string { |
| + if rest, ok := strings.CutPrefix(target, "accounts/"); ok { |
| + if a, _, ok := strings.Cut(rest, "/"); ok && a != "" { |
| + return "accounts/" + a |
| + } |
| + return "" |
| + } |
| + if sc, ok := release.ChangeScope(target); ok { |
| + return sc |
| + } |
| + return "" |
| +} |
| + |
| +// read runs before every read the view makes: it attributes the read to the |
| +// operation in ctx, fires that operation's triggers for it, and fails it if |
| +// the plan says so. Reads outside an operation (the checks) pass untouched. |
| +func (s *sim) read(ctx context.Context, target string) error { |
| + op, _ := ctx.Value(simOpKey{}).(*simOp) |
| + if op == nil { |
| + return nil |
| + } |
| + s.mu.Lock() |
| + if target == release.ChangeIndexKey { |
| + if !op.polled { |
| + op.polled, op.pollSeq = true, s.done |
| + } |
| + fail := op.fail[target] |
| + if fail { |
| + op.failedAny = true |
| + s.stats.readFailures++ |
| + } |
| + s.mu.Unlock() |
| + if fail { |
| + return errSimRead |
| + } |
| + return nil |
| + } |
| + if target == "accounts/" { |
| + op.walked = true |
| + } |
| + scope := simScope(target) |
| + var fire []*simTrigger |
| + if scope != "" { |
| + n := op.counts[scope] |
| + op.counts[scope] = n + 1 |
| + for _, tr := range op.triggers { |
| + if !tr.fired && tr.scope == scope && tr.n == n { |
| + tr.fired = true |
| + fire = append(fire, tr) |
| + } |
| + } |
| + } |
| + s.mu.Unlock() |
| + for _, tr := range fire { |
| + s.tracef(" %s: before read %d of %s (%s): %s", op.kind, tr.n, scope, target, tr.what) |
| + s.mu.Lock() |
| + s.stats.triggers++ |
| + s.mu.Unlock() |
| + tr.act(op) |
| + } |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + fail := op.fail[target] |
| + if scope != "" { |
| + op.read[scope] = true |
| + if fail { |
| + op.failed[scope] = true |
| + } |
| + } |
| + if fail { |
| + op.failedAny = true |
| + s.stats.readFailures++ |
| + return errSimRead |
| + } |
| + return nil |
| +} |
| + |
| +// simStore is the bucket as the view reads it. |
| +type simStore struct { |
| + *release.DirStore |
| + bucket bool |
| + sim *sim |
| +} |
| + |
| +func (s *simStore) BucketShapedListDirs() bool { return s.bucket } |
| + |
| +func (s *simStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if err := s.sim.read(ctx, key); err != nil { |
| + return nil, "", err |
| + } |
| + return s.DirStore.Get(ctx, key) |
| +} |
| + |
| +func (s *simStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| + if err := s.sim.read(ctx, key); err != nil { |
| + return nil, "", false, err |
| + } |
| + return s.DirStore.GetIfChanged(ctx, key, etag) |
| +} |
| + |
| +func (s *simStore) List(ctx context.Context, prefix string) ([]string, error) { |
| + if err := s.sim.read(ctx, prefix); err != nil { |
| + return nil, err |
| + } |
| + return s.DirStore.List(ctx, prefix) |
| +} |
| + |
| +func (s *simStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| + if err := s.sim.read(ctx, prefix); err != nil { |
| + return nil, err |
| + } |
| + return s.DirStore.ListETags(ctx, prefix) |
| +} |
| + |
| +func (s *simStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| + if err := s.sim.read(ctx, prefix); err != nil { |
| + return nil, err |
| + } |
| + return s.DirStore.ListDirs(ctx, prefix) |
| +} |
| + |
| +func (s *simStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| + if err := s.sim.read(ctx, prefix); err != nil { |
| + return nil, err |
| + } |
| + return s.DirStore.ListMod(ctx, prefix) |
| +} |
| + |
| +// ---- writers ---- |
| + |
| +func (s *sim) indexGen(scope string) uint64 { |
| + idx, _, err := release.ReadChangeIndex(ctx, s.raw) |
| + if err != nil { |
| + s.fail("sim: reading the index: %v", err) |
| + return 0 |
| + } |
| + return idx.Entries[scope] |
| +} |
| + |
| +func (s *sim) body(w simWrite, seq int) []byte { |
| + var v any |
| + if a, ok := domains.AccountManifestSqid(w.key); ok { |
| + m := domains.AccountManifest{ID: s.acctID[a], Sqid: a, Name: fmt.Sprintf("Account v%d", seq), CreatedAt: simEpoch} |
| + if w.soft { |
| + m.DeletedAt = simEpoch.Add(time.Hour) |
| + } |
| + v = m |
| + } else { |
| + a, b, rest, _ := appKey(w.key) |
| + switch { |
| + case rest == "instances.json": |
| + v = Manifest{Schema: 1, App: b, Domain: fmt.Sprintf("v%d.%s.example.com", seq, b)} |
| + case rest == "sidecars.json": |
| + v = SidecarManifest{Schema: 1, App: b, Args: []string{"run"}, Channel: fmt.Sprintf("c%d", seq)} |
| + case rest == "deleted.json": |
| + v = release.AppTombstone{Schema: 1, Account: a, App: b, DeletedAt: simEpoch, Actor: fmt.Sprintf("t%d", seq)} |
| + case strings.HasPrefix(rest, "instances/"): |
| + host := strings.TrimSuffix(strings.TrimPrefix(rest, "instances/"), ".json") |
| + slug, _, _ := strings.Cut(host, ".") |
| + v = Record{Schema: 1, Host: host, Slug: slug, App: b, Account: a, Region: fmt.Sprintf("r%d", seq), Channel: "stable", CreatedAt: simEpoch} |
| + case strings.HasPrefix(rest, "sidecars/"): |
| + name := strings.TrimSuffix(strings.TrimPrefix(rest, "sidecars/"), ".json") |
| + v = SidecarRecord{Schema: 1, Name: name, App: b, Account: a, Host: SidecarHost(name, b, a), Channel: fmt.Sprintf("c%d", seq)} |
| + default: |
| + panic("sim: no body for " + w.key) |
| + } |
| + } |
| + body, err := json.Marshal(v) |
| + if err != nil { |
| + panic(err) |
| + } |
| + return body |
| +} |
| + |
| +// begin registers a write before it touches the bucket, so a reader that |
| +// sees it can always name it. Caller holds writeMu. |
| +func (s *sim) begin(w simWrite) (seq int, body []byte, etag string, before uint64) { |
| + scope, _ := release.ChangeScope(w.key) |
| + before = s.indexGen(scope) |
| + if _, e, err := s.raw.Get(ctx, w.key); err == nil { |
| + etag = e |
| + } |
| + s.mu.Lock() |
| + s.seq++ |
| + seq = s.seq |
| + s.mu.Unlock() |
| + if !w.del { |
| + body = s.body(w, seq) |
| + } |
| + s.mu.Lock() |
| + s.hist[w.key] = append(s.hist[w.key], simVersion{seq: seq, body: body}) |
| + s.mu.Unlock() |
| + if w.w.failBump { |
| + _ = os.Chmod(filepath.Join(s.root, "control"), 0o555) |
| + } |
| + return seq, body, etag, before |
| +} |
| + |
| +// end classifies the write by what the index did, and completes it. |
| +func (s *sim) end(w simWrite, seq int, before uint64, err error) { |
| + if w.w.failBump { |
| + _ = os.Chmod(filepath.Join(s.root, "control"), 0o755) |
| + } |
| + if err != nil { |
| + s.fail("sim: %s write of %s: %v", w.w.name, w.key, err) |
| + } |
| + scope, _ := release.ChangeScope(w.key) |
| + after := s.indexGen(scope) |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.done = seq |
| + s.stats.writes++ |
| + s.dirty[scope] = true |
| + class := "no bump" |
| + switch { |
| + case after == before+1: |
| + s.bumped[w.key] = append(s.bumped[w.key], seq) |
| + s.stats.bumped++ |
| + class = fmt.Sprintf("bumped %s to %d", scope, after) |
| + case after != before: |
| + s.errs = append(s.errs, fmt.Sprintf("sim: write %d moved %s from %d to %d", seq, scope, before, after)) |
| + case w.w.failBump: |
| + s.stats.failedBumps++ |
| + default: |
| + s.stats.bypassed++ |
| + } |
| + op := "put" |
| + if w.del { |
| + op = "delete" |
| + } |
| + s.trace = append(s.trace, fmt.Sprintf("[%d] write %d: %s %s %s (%s)", s.step, seq, w.w.name, op, w.key, class)) |
| +} |
| + |
| +func (s *sim) write(w simWrite) { |
| + s.writeMu.Lock() |
| + defer s.writeMu.Unlock() |
| + seq, body, etag, before := s.begin(w) |
| + var err error |
| + if w.del { |
| + err = w.w.store.Delete(ctx, w.key) |
| + } else { |
| + _, err = w.w.store.Put(ctx, w.key, body, etag) |
| + } |
| + s.end(w, seq, before, err) |
| +} |
| + |
| +// simGate is a request body that holds the write at its first Read until |
| +// released: the write has begun and is not yet in the bucket. |
| +type simGate struct { |
| + r *bytes.Reader |
| + started chan struct{} |
| + release chan struct{} |
| + once sync.Once |
| +} |
| + |
| +func (g *simGate) Read(p []byte) (int, error) { |
| + g.once.Do(func() { |
| + close(g.started) |
| + <-g.release |
| + }) |
| + return g.r.Read(p) |
| +} |
| + |
| +func (g *simGate) Seek(off int64, whence int) (int64, error) { return g.r.Seek(off, whence) } |
| + |
| +// pausedWrite streams a put whose body stalls, runs view operations while |
| +// it is in flight, then lets it finish and runs the next pass. |
| +func (s *sim) pausedWrite(w simWrite, during func()) { |
| + s.writeMu.Lock() |
| + seq, body, etag, before := s.begin(w) |
| + s.tracef(" write %d in flight: %s put %s", seq, w.w.name, w.key) |
| + g := &simGate{r: bytes.NewReader(body), started: make(chan struct{}), release: make(chan struct{})} |
| + done := make(chan error, 1) |
| + go func() { |
| + _, err := w.w.store.PutFrom(ctx, w.key, g, int64(len(body)), etag) |
| + done <- err |
| + }() |
| + var err error |
| + finished := false |
| + select { |
| + case <-g.started: |
| + case err = <-done: |
| + finished = true |
| + } |
| + if !finished { |
| + s.paused, s.writes = true, false |
| + s.mu.Lock() |
| + s.stats.paused++ |
| + s.mu.Unlock() |
| + during() |
| + s.paused, s.writes = false, true |
| + close(g.release) |
| + err = <-done |
| + } |
| + s.end(w, seq, before, err) |
| + s.writeMu.Unlock() |
| +} |
| + |
| +// ---- view operations ---- |
| + |
| +// run is one view operation and the checks that follow every operation. |
| +func (s *sim) run(op *simOp) error { |
| + cctx := context.WithValue(context.Background(), simOpKey{}, op) |
| + var err error |
| + if op.kind == "refresh" { |
| + err = s.v.Refresh(cctx) |
| + } else { |
| + err = s.v.Rebuild(cctx) |
| + } |
| + s.mu.Lock() |
| + op.err = err |
| + if op.kind == "refresh" { |
| + s.stats.refreshes++ |
| + } else { |
| + s.stats.rebuilds++ |
| + } |
| + s.mu.Unlock() |
| + sn := s.v.Snapshot() |
| + var ep uint64 |
| + complete := false |
| + if sn != nil { |
| + ep, complete = sn.Epoch, sn.Complete |
| + } |
| + var fails []string |
| + for k := range op.fail { |
| + fails = append(fails, k) |
| + } |
| + sort.Strings(fails) |
| + s.tracef(" %s%s: err=%v epoch=%d complete=%v polled-at-write=%d failing=%v", map[bool]string{true: "nested "}[op.nested], op.kind, err, ep, complete, op.pollSeq, fails) |
| + if op.kind == "refresh" && err == nil && op.polled { |
| + s.checkVisible(op) |
| + } |
| + s.checkSnapshot() |
| + return err |
| +} |
| + |
| +// top runs an operation as a pass would, then the (b) and (e) checks, then |
| +// any concurrent Refresh that had to wait for mu. |
| +func (s *sim) top(op *simOp) error { |
| + carried := make(map[string]bool, len(s.carry)) |
| + for sc := range s.carry { |
| + carried[sc] = true |
| + } |
| + err := s.run(op) |
| + |
| + // (e): an incremental Refresh reads again every scope whose last read |
| + // failed. A Refresh that walked read the whole deployment instead. |
| + if op.kind == "refresh" && !op.walked { |
| + for sc := range carried { |
| + s.mu.Lock() |
| + s.stats.checksE++ |
| + s.mu.Unlock() |
| + if !op.read[sc] { |
| + s.fail("(e) %s failed its last read and this Refresh did not read it again", sc) |
| + } |
| + } |
| + } |
| + if op.walked && err == nil { |
| + // A walk that got past its top-level listings read the whole |
| + // deployment: a scope it did not reach has no keys left to read. |
| + clear(s.carry) |
| + } |
| + for sc := range op.read { |
| + if !op.failed[sc] { |
| + delete(s.carry, sc) |
| + } |
| + } |
| + for sc := range op.failed { |
| + s.carry[sc] = true |
| + } |
| + |
| + if op.kind == "rebuild" && err == nil && !op.failedAny && !op.wrote && !op.nestedRan && !s.paused { |
| + s.checkMatchesLive() |
| + } |
| + |
| + for { |
| + s.mu.Lock() |
| + n := s.deferred |
| + s.deferred = 0 |
| + s.mu.Unlock() |
| + if n == 0 { |
| + break |
| + } |
| + for range n { |
| + s.mu.Lock() |
| + s.stats.deferred++ |
| + s.mu.Unlock() |
| + s.top(newSimOp("refresh")) |
| + } |
| + } |
| + return err |
| +} |
| + |
| +// concurrent is a Refresh started by another caller at this instant. It gets |
| +// in only if mu is free; otherwise it waits for the operation holding mu. |
| +func (s *sim) concurrent(outer *simOp) { |
| + if s.v.mu.TryLock() { |
| + s.v.mu.Unlock() |
| + s.mu.Lock() |
| + outer.nestedRan = true |
| + s.stats.nested++ |
| + s.mu.Unlock() |
| + op := newSimOp("refresh") |
| + op.nested = true |
| + s.run(op) |
| + return |
| + } |
| + s.mu.Lock() |
| + s.deferred++ |
| + s.mu.Unlock() |
| +} |
| + |
| +func (s *sim) siblingPoll() { |
| + s.mu.Lock() |
| + s.stats.sibling++ |
| + s.mu.Unlock() |
| + _, _ = s.watch.Poll(context.Background()) |
| +} |
| + |
| +// ---- checks ---- |
| + |
| +// version is the newest write whose value the answer shows: a body names |
| +// its write uniquely, and an absence is credited to the newest delete. |
| +func (s *sim) version(key string, body []byte, absent bool) (int, bool) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + best := -1 |
| + for _, v := range s.hist[key] { |
| + if (absent && v.body == nil) || (!absent && v.body != nil && bytes.Equal(v.body, body)) { |
| + best = max(best, v.seq) |
| + } |
| + } |
| + return best, best >= 0 |
| +} |
| + |
| +func (s *sim) answer(st release.Store, key string) (body []byte, absent, failed bool) { |
| + body, _, err := st.Get(context.Background(), key) |
| + switch { |
| + case err == nil: |
| + return body, false, false |
| + case errors.Is(err, os.ErrNotExist): |
| + return nil, true, false |
| + } |
| + return nil, false, true |
| +} |
| + |
| +// checkVisible is (a). |
| +func (s *sim) checkVisible(op *simOp) { |
| + sn := s.v.Snapshot() |
| + if sn == nil { |
| + s.fail("(a) a Refresh returned nil and published nothing") |
| + return |
| + } |
| + st := sn.Store(s.live) |
| + s.mu.Lock() |
| + need := map[string]int{} |
| + for key, seqs := range s.bumped { |
| + for _, q := range seqs { |
| + if q <= op.pollSeq { |
| + need[key] = max(need[key], q) |
| + } |
| + } |
| + } |
| + s.mu.Unlock() |
| + for key, q := range need { |
| + body, absent, failed := s.answer(st, key) |
| + if failed { |
| + continue // a cached read failure: no consumer acts on it |
| + } |
| + ver, ok := s.version(key, body, absent) |
| + switch { |
| + case !ok: |
| + s.fail("(a) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body) |
| + case ver < q: |
| + s.fail("(a) epoch %d: Refresh polled after write %d (bumped) completed, but %s shows write %d", sn.Epoch, q, key, ver) |
| + } |
| + } |
| +} |
| + |
| +// checkSnapshot is (c) and (d) over the current snapshot, once per epoch. |
| +func (s *sim) checkSnapshot() { |
| + s.checkMu.Lock() |
| + defer s.checkMu.Unlock() |
| + sn := s.v.Snapshot() |
| + if sn == nil { |
| + return |
| + } |
| + if sn.Epoch < s.lastEp { |
| + s.fail("(c) published epoch went from %d back to %d", s.lastEp, sn.Epoch) |
| + return |
| + } |
| + if sn.Epoch == s.lastEp { |
| + return |
| + } |
| + s.lastEp = sn.Epoch |
| + if !sn.Complete { |
| + s.mu.Lock() |
| + s.stats.incomplete++ |
| + s.mu.Unlock() |
| + } |
| + st := sn.Store(s.live) |
| + bg := context.Background() |
| + famErr := "" |
| + for _, key := range s.keys { |
| + body, absent, failed := s.answer(st, key) |
| + if failed { |
| + if famErr == "" { |
| + famErr = "Get " + key |
| + } |
| + continue |
| + } |
| + ver, ok := s.version(key, body, absent) |
| + if !ok { |
| + s.fail("(c) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body) |
| + continue |
| + } |
| + if ver < s.lastVer[key] { |
| + s.fail("(c) epoch %d: %s went back from write %d to write %d", sn.Epoch, key, s.lastVer[key], ver) |
| + } |
| + s.lastVer[key] = ver |
| + } |
| + for _, a := range s.acct { |
| + if _, err := st.List(bg, "accounts/"+a+"/"); err != nil && famErr == "" { |
| + famErr = "List accounts/" + a + "/" |
| + } |
| + if _, err := st.ListDirs(bg, "apps/"+a+"/"); err != nil && famErr == "" { |
| + famErr = "ListDirs apps/" + a + "/" |
| + } |
| + for _, b := range simApps { |
| + for _, p := range []string{RecordsPrefix(a, b), SidecarsPrefix(a, b)} { |
| + if _, err := st.List(bg, p); err != nil && famErr == "" { |
| + famErr = "List " + p |
| + } |
| + } |
| + } |
| + } |
| + if _, rf, err := knownAccounts(bg, st); (err != nil || rf) && famErr == "" { |
| + famErr = "knownAccounts" |
| + } |
| + _, _, strictErr := ListAllRecordsStrict(bg, st) |
| + switch { |
| + case sn.Complete && strictErr != nil: |
| + s.fail("(d) epoch %d is Complete but ListAllRecordsStrict over it fails: %v", sn.Epoch, strictErr) |
| + case sn.Complete && famErr != "": |
| + s.fail("(d) epoch %d is Complete but %s over it fails", sn.Epoch, famErr) |
| + case !sn.Complete && strictErr == nil && recordWalkGap(sn): |
| + s.fail("(d) epoch %d is not Complete, its gap is in the record walk, and ListAllRecordsStrict over it returned nil", sn.Epoch) |
| + case !sn.Complete && strictErr == nil && famErr == "": |
| + s.fail("(d) epoch %d is not Complete, yet every reader over it answers as if it were", sn.Epoch) |
| + } |
| +} |
| + |
| +// recordWalkGap reports a gap in what ListAllRecordsStrict reads: the |
| +// top-level walk, an account it would know (or fail to know), a known |
| +// account's app directories, an instances listing or a record body. |
| +func recordWalkGap(sn *ViewSnapshot) bool { |
| + if sn.walkErr != nil { |
| + return true |
| + } |
| + known, readFailed, err := knownAccounts(context.Background(), sn.Store(uncachedStore{})) |
| + if err != nil || readFailed { |
| + return true |
| + } |
| + for a := range known { |
| + if sn.dirErrs[a] != nil { |
| + return true |
| + } |
| + for _, e := range sn.apps[a] { |
| + if e.instErr != nil { |
| + return true |
| + } |
| + for _, ke := range e.inst { |
| + if e.objs[ke.Key].failed() { |
| + return true |
| + } |
| + } |
| + } |
| + } |
| + return false |
| +} |
| + |
| +// checkMatchesLive is (b): after a clean Rebuild with nothing written during |
| +// it, the readers over the snapshot answer exactly as over the bucket. |
| +func (s *sim) checkMatchesLive() { |
| + s.stats.checksB++ |
| + sn := s.v.Snapshot() |
| + if !sn.Complete { |
| + s.fail("(b) a clean Rebuild published epoch %d not Complete", sn.Epoch) |
| + return |
| + } |
| + lr, lu, lerr := ListAllRecordsStrict(ctx, s.clean) |
| + sr, su, serr := ListAllRecordsStrict(ctx, sn.Store(s.live)) |
| + if lerr != nil || serr != nil { |
| + s.fail("(b) ListAllRecordsStrict: bucket %v, snapshot %v", lerr, serr) |
| + return |
| + } |
| + norm := func(rs []Record) []Record { |
| + for i := range rs { |
| + rs[i].CreatedAt = rs[i].CreatedAt.UTC() |
| + } |
| + return rs |
| + } |
| + if !reflect.DeepEqual(norm(lr), norm(sr)) || !reflect.DeepEqual(lu, su) { |
| + s.fail("(b) ListAllRecordsStrict over epoch %d:\n%+v %v\nover the bucket:\n%+v %v", sn.Epoch, sr, su, lr, lu) |
| + return |
| + } |
| + if want, got := readWalk(s.t, s.clean, s.f), readWalk(s.t, sn.Store(s.live), s.f); !reflect.DeepEqual(got, want) { |
| + s.fail("(b) the record walk over epoch %d:\n%+v\nover the bucket:\n%+v", sn.Epoch, got, want) |
| + } |
| +} |
| + |
| +// ---- the consumer ---- |
| + |
| +// consumerBegin is a pass's Reconcile leg: refresh, take the snapshot, and |
| +// hand it over towards w.pass. |
| +func (s *sim) consumerBegin(op *simOp) { |
| + s.top(op) |
| + if sn := s.v.Snapshot(); sn != nil { |
| + s.hand = append(s.hand, simHandover{sn: sn, from: s.step}) |
| + s.tracef(" pass begins with epoch %d handed over", sn.Epoch) |
| + } |
| +} |
| + |
| +// consumerFinish is the moves pass reaching w.pass: it refreshes and takes |
| +// the current snapshot under it, records the epoch it consumed and refuses a |
| +// lower one. simConsumeHandedOver makes it use the handed-over snapshot. |
| +func (s *sim) consumerFinish(i int, op *simOp) { |
| + h := s.hand[i] |
| + s.hand = slices.Delete(s.hand, i, i+1) |
| + s.pass.Lock() |
| + defer s.pass.Unlock() |
| + sn, err := h.sn, error(nil) |
| + if !simConsumeHandedOver { |
| + err = s.top(op) |
| + sn = s.v.Snapshot() |
| + } |
| + if sn == nil { |
| + return |
| + } |
| + s.tracef(" moves pass (begun at step %d) consumes epoch %d (complete=%v, err=%v)", h.from, sn.Epoch, sn.Complete, err) |
| + s.stats.consumed++ |
| + if sn.Epoch < s.consumed { |
| + s.fail("(c) the moves pass consumed epoch %d after it had consumed %d", sn.Epoch, s.consumed) |
| + } |
| + s.consumed = max(s.consumed, sn.Epoch) |
| + if err == nil && sn.Complete { |
| + if _, _, err := ListAllRecordsStrict(context.Background(), sn.Store(s.live)); err != nil { |
| + s.fail("(d) the moves pass acted on Complete epoch %d and the strict reader failed: %v", sn.Epoch, err) |
| + } |
| + } |
| +} |
| + |
| +// ---- the generator ---- |
| + |
| +func (s *sim) pickWriter() *simWriter { |
| + switch r := s.rng.IntN(10); { |
| + case r < 6: |
| + return s.bumpW |
| + case r < 8: |
| + return s.bypW |
| + } |
| + return s.failW |
| +} |
| + |
| +func (s *sim) pickScope() string { |
| + a := s.acct[s.rng.IntN(len(s.acct))] |
| + if s.rng.IntN(100) < 15 { |
| + return "accounts/" + a |
| + } |
| + return "apps/" + a + "/" + simApps[s.rng.IntN(len(simApps))] |
| +} |
| + |
| +func (s *sim) pickWrite(scope string, w *simWriter, puts bool) simWrite { |
| + if a, ok := scopeAccount(scope); ok { |
| + return simWrite{w: w, key: accountManifestKey(a), soft: s.rng.IntN(100) < 15} |
| + } |
| + a, b, _ := ScopeApp(scope) |
| + slug := simSlugs[s.rng.IntN(len(simSlugs))] |
| + name := simSidecars[s.rng.IntN(len(simSidecars))] |
| + r := s.rng.IntN(20) |
| + if puts && r >= 8 && r < 11 { |
| + r = 0 |
| + } |
| + switch { |
| + case r < 8: |
| + return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com")} |
| + case r < 11: |
| + return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com"), del: true} |
| + case r < 14: |
| + return simWrite{w: w, key: SidecarRecordKey(a, b, name), del: !puts && r == 13} |
| + case r < 16: |
| + return simWrite{w: w, key: ManifestKey(a, b)} |
| + case r < 17: |
| + return simWrite{w: w, key: SidecarManifestKey(a, b)} |
| + } |
| + return simWrite{w: w, key: release.AppTombstoneKey(a, b), del: !puts && r == 19} |
| +} |
| + |
| +func (s *sim) pickFailTarget() string { |
| + a := s.acct[s.rng.IntN(len(s.acct))] |
| + b := simApps[s.rng.IntN(len(simApps))] |
| + switch r := s.rng.IntN(100); { |
| + case r < 70: |
| + slug := simSlugs[s.rng.IntN(len(simSlugs))] |
| + return []string{ |
| + RecordsPrefix(a, b), SidecarsPrefix(a, b), RecordKey(a, b, slug+"."+b+".example.com"), |
| + SidecarRecordKey(a, b, simSidecars[s.rng.IntN(len(simSidecars))]), |
| + ManifestKey(a, b), SidecarManifestKey(a, b), release.AppTombstoneKey(a, b), |
| + }[s.rng.IntN(7)] |
| + case r < 85: |
| + return []string{"accounts/" + a + "/", accountManifestKey(a)}[s.rng.IntN(2)] |
| + } |
| + return []string{"accounts/", "apps/", "apps/" + a + "/", release.ChangeIndexKey}[s.rng.IntN(4)] |
| +} |
| + |
| +// triggerScope picks where a trigger waits. An incremental Refresh reads only |
| +// the scopes that moved or failed, so those are favoured; a Rebuild reads |
| +// every scope. |
| +func (s *sim) triggerScope(kind string) string { |
| + s.mu.Lock() |
| + var hot []string |
| + for sc := range s.dirty { |
| + hot = append(hot, sc) |
| + } |
| + s.mu.Unlock() |
| + for sc := range s.carry { |
| + if !slices.Contains(hot, sc) { |
| + hot = append(hot, sc) |
| + } |
| + } |
| + sort.Strings(hot) |
| + if kind == "refresh" && len(hot) > 0 && s.rng.IntN(100) < 70 { |
| + return hot[s.rng.IntN(len(hot))] |
| + } |
| + return s.pickScope() |
| +} |
| + |
| +// plan draws an operation: read failures, and triggers that write into the |
| +// scope being read, start a concurrent Refresh, or poll the shared watch. |
| +func (s *sim) plan(kind string) *simOp { |
| + op := newSimOp(kind) |
| + if s.rng.IntN(100) < 30 { |
| + for range 1 + s.rng.IntN(2) { |
| + op.fail[s.pickFailTarget()] = true |
| + } |
| + } |
| + defer func() { |
| + s.mu.Lock() |
| + clear(s.dirty) |
| + s.mu.Unlock() |
| + }() |
| + for range s.rng.IntN(4) { |
| + tr := &simTrigger{scope: s.triggerScope(kind), n: s.rng.IntN(6)} |
| + k := s.rng.IntN(20) |
| + if !s.writes && k < 15 { |
| + k = 15 + s.rng.IntN(5) |
| + } |
| + switch { |
| + case k < 12: |
| + wr := s.pickWrite(tr.scope, s.pickWriter(), false) |
| + tr.what = fmt.Sprintf("%s writes %s", wr.w.name, wr.key) |
| + tr.act = func(o *simOp) { s.hookWrite(o, wr) } |
| + case k < 15: |
| + // The codex-2 shape: a write lands behind the reader, and a |
| + // concurrent Refresh starts at once. |
| + wr := s.pickWrite(tr.scope, s.bumpW, true) |
| + tr.what = fmt.Sprintf("bumping writes %s, then a concurrent Refresh", wr.key) |
| + tr.act = func(o *simOp) { s.hookWrite(o, wr); s.concurrent(o) } |
| + case k < 18: |
| + tr.what = "a concurrent Refresh" |
| + tr.act = s.concurrent |
| + default: |
| + tr.what = "a sibling polls the shared watch" |
| + tr.act = func(*simOp) { s.siblingPoll() } |
| + } |
| + op.triggers = append(op.triggers, tr) |
| + } |
| + return op |
| +} |
| + |
| +func (s *sim) hookWrite(o *simOp, wr simWrite) { |
| + s.mu.Lock() |
| + o.wrote = true |
| + s.mu.Unlock() |
| + s.write(wr) |
| +} |
| + |
| +// viewStep draws one view-side step: a Refresh, a Rebuild, or one leg of a |
| +// moves pass. |
| +func (s *sim) viewStep() { |
| + switch r := s.rng.IntN(100); { |
| + case r < 45: |
| + s.tracef("refresh") |
| + s.top(s.plan("refresh")) |
| + case r < 60: |
| + s.tracef("rebuild") |
| + s.top(s.plan("rebuild")) |
| + case r < 75: |
| + s.tracef("pass begins") |
| + s.consumerBegin(s.plan("refresh")) |
| + case r < 90 && len(s.hand) > 0: |
| + s.tracef("moves pass reaches w.pass") |
| + s.consumerFinish(s.rng.IntN(len(s.hand)), s.plan("refresh")) |
| + default: |
| + s.tracef("quiet refresh") |
| + s.top(newSimOp("refresh")) |
| + } |
| +} |
| + |
| +func (s *sim) randomStep() { |
| + switch r := s.rng.IntN(100); { |
| + case r < 22: |
| + wr := s.pickWrite(s.pickScope(), s.pickWriter(), false) |
| + s.tracef("write") |
| + s.write(wr) |
| + case r < 30: |
| + w := s.bumpW |
| + if s.rng.IntN(10) < 3 { |
| + w = s.pickWriter() |
| + } |
| + wr := s.pickWrite(s.pickScope(), w, true) |
| + s.tracef("write in flight") |
| + s.pausedWrite(wr, func() { |
| + s.tracef("refresh while a write is in flight") |
| + s.top(s.plan("refresh")) |
| + if s.rng.IntN(2) == 0 { |
| + s.viewStep() |
| + } |
| + }) |
| + s.tracef("refresh after the write completed") |
| + s.top(s.plan("refresh")) |
| + case r < 33: |
| + s.tracef("sibling poll") |
| + s.siblingPoll() |
| + default: |
| + s.viewStep() |
| + } |
| +} |
| + |
| +func runSimSeed(t *testing.T, tmpl *simTemplate, seed uint64) simStats { |
| + s := newSim(t, tmpl, seed, seed%2 == 0) |
| + steps := 20 + s.rng.IntN(15) |
| + for s.step = 1; s.step <= steps; s.step++ { |
| + s.randomStep() |
| + s.stats.steps++ |
| + s.stop() |
| + } |
| + // Quiesce: a clean backstop, which (b) checks, then a clean refresh. |
| + s.tracef("final rebuild") |
| + s.top(newSimOp("rebuild")) |
| + s.tracef("final refresh") |
| + s.top(newSimOp("refresh")) |
| + s.stop() |
| + return s.stats |
| +} |
| + |
| +// ---- the named interleavings (§8) ---- |
| + |
| +func simScripted(t *testing.T, tmpl *simTemplate) { |
| + for _, bucket := range []bool{true, false} { |
| + // codex 2: the backstop lists an app before a record appears, and an |
| + // incremental refresh sees the record. The refresh must wait for the |
| + // backstop and publish after it, so the record is never published |
| + // and then lost. |
| + t.Run(fmt.Sprintf("codex2-backstop-vs-refresh/bucket=%v", bucket), func(t *testing.T) { |
| + s := newSim(t, tmpl, 1, bucket) |
| + s.top(newSimOp("rebuild")) |
| + a := s.acct[0] |
| + key := RecordKey(a, "shop", "three.shop.example.com") |
| + op := newSimOp("rebuild") |
| + op.triggers = []*simTrigger{{scope: "apps/" + a + "/shop", n: 1, what: "record appears, concurrent Refresh", |
| + act: func(o *simOp) { s.hookWrite(o, simWrite{w: s.bumpW, key: key}); s.concurrent(o) }}} |
| + s.top(op) |
| + s.stop() |
| + if !op.triggers[0].fired { |
| + t.Fatal("the trigger never fired") |
| + } |
| + if body, absent, failed := s.answer(s.v.Snapshot().Store(s.live), key); absent || failed || body == nil { |
| + t.Fatalf("the record is missing after the backstop and the refresh behind it (absent %v, failed %v)", absent, failed) |
| + } |
| + }) |
| + // codex 4: an app's re-read fails after the index moved, and the |
| + // next index read is a 304. The next Refresh must read it again. |
| + t.Run(fmt.Sprintf("codex4-failed-reread-then-304/bucket=%v", bucket), func(t *testing.T) { |
| + s := newSim(t, tmpl, 1, bucket) |
| + s.top(newSimOp("rebuild")) |
| + a := s.acct[0] |
| + key := RecordKey(a, "shop", "three.shop.example.com") |
| + s.write(simWrite{w: s.bumpW, key: key}) |
| + op := newSimOp("refresh") |
| + op.fail[RecordsPrefix(a, "shop")] = true |
| + s.top(op) |
| + if !op.failed["apps/"+a+"/shop"] { |
| + t.Fatal("the re-read did not fail") |
| + } |
| + _, before, _ := s.raw.Get(ctx, release.ChangeIndexKey) |
| + again := newSimOp("refresh") |
| + s.top(again) |
| + s.stop() |
| + if _, after, _ := s.raw.Get(ctx, release.ChangeIndexKey); after != before { |
| + t.Fatal("the index moved: the second poll was not a 304") |
| + } |
| + if !again.read["apps/"+a+"/shop"] { |
| + t.Fatal("the failed scope was not read again") |
| + } |
| + sn := s.v.Snapshot() |
| + if body, absent, failed := s.answer(sn.Store(s.live), key); absent || failed || body == nil || !sn.Complete { |
| + t.Fatalf("after the retry: absent %v, failed %v, complete %v", absent, failed, sn.Complete) |
| + } |
| + }) |
| + // codex 1: a pass captures a snapshot, stalls; another pass consumes |
| + // a newer one; the first then reaches w.pass. It must not consume an |
| + // older epoch than the second did. |
| + t.Run(fmt.Sprintf("codex1-stalled-pass/bucket=%v", bucket), func(t *testing.T) { |
| + s := newSim(t, tmpl, 1, bucket) |
| + s.top(newSimOp("rebuild")) |
| + s.consumerBegin(newSimOp("refresh")) |
| + s.write(simWrite{w: s.bumpW, key: RecordKey(s.acct[1], "docs", "three.docs.example.com")}) |
| + s.consumerBegin(newSimOp("refresh")) |
| + if s.hand[1].sn.Epoch <= s.hand[0].sn.Epoch { |
| + t.Fatalf("fixture: handovers at epochs %d and %d", s.hand[0].sn.Epoch, s.hand[1].sn.Epoch) |
| + } |
| + s.consumerFinish(1, newSimOp("refresh")) |
| + s.consumerFinish(0, newSimOp("refresh")) |
| + s.stop() |
| + }) |
| + } |
| +} |
| + |
| +func TestRecordViewSimulation(t *testing.T) { |
| + if testing.Short() { |
| + t.Skip("the invariant simulation is not run under -short") |
| + } |
| + // The view and the bump log every failed read and refused bump; the |
| + // simulation makes thousands on purpose. Serial: it sets the log output. |
| + out := log.Writer() |
| + log.SetOutput(io.Discard) |
| + t.Cleanup(func() { log.SetOutput(out) }) |
| + tmpl := newSimTemplate(t) |
| + |
| + t.Run("scripted", func(t *testing.T) { simScripted(t, tmpl) }) |
| + |
| + first, n := uint64(1), *simSeedsFlag |
| + if *simSeedFlag != 0 { |
| + first, n = *simSeedFlag, 1 |
| + } |
| + var ( |
| + mu sync.Mutex |
| + total simStats |
| + ran int |
| + failSeed uint64 |
| + next atomic.Uint64 |
| + stop atomic.Bool |
| + ) |
| + next.Store(first) |
| + start := time.Now() |
| + t.Run("sweep", func(t *testing.T) { |
| + // A seed spends most of its time waiting for DirStore's fsyncs, so a |
| + // few run at once. Each seed is its own bucket, view and writers; the |
| + // sweep stops starting seeds at the first failure. |
| + var wg sync.WaitGroup |
| + for range simParallel { |
| + wg.Go(func() { |
| + for !stop.Load() { |
| + seed := next.Add(1) - 1 |
| + if seed >= first+uint64(n) { |
| + return |
| + } |
| + var st simStats |
| + ok := t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { st = runSimSeed(t, tmpl, seed) }) |
| + mu.Lock() |
| + if ok { |
| + total.add(st) |
| + ran++ |
| + } else if failSeed == 0 || seed < failSeed { |
| + failSeed = seed |
| + stop.Store(true) |
| + } |
| + mu.Unlock() |
| + } |
| + }) |
| + } |
| + wg.Wait() |
| + if failSeed != 0 { |
| + t.Logf("first failing seed: %d (rerun with -seed %d)", failSeed, failSeed) |
| + } |
| + }) |
| + t.Logf("%d seeds passed in %v: %+v", ran, time.Since(start).Round(time.Millisecond), total) |
| + if t.Failed() || ran < 50 { |
| + return |
| + } |
| + // A simulation that stopped exercising something would pass silently. |
| + for name, c := range map[string]int{ |
| + "bumped writes": total.bumped, "bypassing writes": total.bypassed, "failed bumps": total.failedBumps, |
| + "writes in flight": total.paused, "triggers": total.triggers, "concurrent refreshes that waited": total.deferred, |
| + "sibling polls": total.sibling, "read failures": total.readFailures, "(b) checks": total.checksB, |
| + "(e) checks": total.checksE, "consumed snapshots": total.consumed, "incomplete snapshots": total.incomplete, |
| + "rebuilds": total.rebuilds, |
| + } { |
| + if c == 0 { |
| + t.Errorf("the sweep exercised no %s", name) |
| + } |
| + } |
| +} |