| 1 | package provision |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "encoding/json" |
| 7 | "errors" |
| 8 | "flag" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "log" |
| 12 | "math/rand/v2" |
| 13 | "os" |
| 14 | "path/filepath" |
| 15 | "reflect" |
| 16 | "runtime" |
| 17 | "slices" |
| 18 | "sort" |
| 19 | "strings" |
| 20 | "sync" |
| 21 | "sync/atomic" |
| 22 | "testing" |
| 23 | "time" |
| 24 | |
| 25 | "github.com/carlosframework/platform/internal/domains" |
| 26 | "github.com/carlosframework/platform/internal/release" |
| 27 | ) |
| 28 | |
| 29 | // The RecordView invariant simulation (S3 change-index design §8). Random |
| 30 | // interleavings of writers (bumping, bypassing, failing their bump), |
| 31 | // incremental refreshes, backstop rebuilds and a moves-pass consumer run on a |
| 32 | // DirStore, and five invariants are checked after every step: |
| 33 | // |
| 34 | // (a) after a Refresh that returned nil, every write whose bump succeeded |
| 35 | // and that completed before the Refresh's index poll is in the snapshot |
| 36 | // (or that key answers a read failure, which no consumer acts on); |
| 37 | // (b) after a clean Rebuild (no read failed, no write landed during it), |
| 38 | // the record walk over the snapshot equals the record walk over the |
| 39 | // bucket, ListAllRecordsStrict included; |
| 40 | // (c) published epochs only increase, the consumer never consumes a lower |
| 41 | // epoch than it already consumed, and no key ever reverts to an older |
| 42 | // write; |
| 43 | // (d) a snapshot with Complete == false never answers ListAllRecordsStrict |
| 44 | // with a nil error when its gap lies in what that reader reads (an |
| 45 | // account, an instances listing, a record body); a gap anywhere else |
| 46 | // (a sidecar listing, a manifest) is still visible as a read error to |
| 47 | // that family's reader; and a Complete snapshot answers every reader |
| 48 | // without error; |
| 49 | // (e) a scope whose read failed is read again by the next incremental |
| 50 | // Refresh, whether or not the index moved. |
| 51 | // |
| 52 | // How the interleavings are made. The view reads through simStore, which |
| 53 | // carries the current operation in its ctx and calls the simulation before |
| 54 | // every read: a read may fail by the operation's plan, and a trigger planned |
| 55 | // for "before the n-th read of scope S" fires there, writing to S, polling |
| 56 | // the shared watch as a sibling consumer would, or starting a concurrent |
| 57 | // Refresh. The concurrent Refresh runs nested, right there, if RecordView.mu |
| 58 | // is free (TryLock), which is exactly when a real concurrent caller would |
| 59 | // get in; otherwise it would block on mu, so it is queued and runs after the |
| 60 | // operation returns. Triggers fire in the reading goroutine, before the read, |
| 61 | // and only write to the scope being read, so where a write lands relative to |
| 62 | // that scope's reads does not depend on goroutine timing. |
| 63 | // |
| 64 | // A write that bumps and is still in flight (the one interleaving a store |
| 65 | // call cannot expose: between a write and its bump) is a DirStore.PutFrom |
| 66 | // whose body reader blocks on its first Read; refreshes run while it waits. |
| 67 | // A write whose bump fails is a real index-enabled DirStore whose own bump |
| 68 | // is refused through the production path: while its write is in progress |
| 69 | // the index file has a second hard link (outside the bucket), and safefile |
| 70 | // refuses to open any file with more than one link, for every user, root |
| 71 | // included. The bump's read of the index fails, so the bump gives up and the |
| 72 | // write stands. No other reader runs at that instant except a sibling poll |
| 73 | // fired from another reading goroutine, which then finds the index |
| 74 | // unusable and makes the next Refresh walk, as it should. Whether a write |
| 75 | // bumped is never assumed: the index is read before and after each write |
| 76 | // (writes are serialized), and the write is classed by what happened. |
| 77 | // |
| 78 | // The three interleavings §8 names are scripted subtests that run before the |
| 79 | // sweep, under both ListDirs shapes: codex 2 (the backstop has listed an app |
| 80 | // when a record appears and a concurrent Refresh starts), codex 4 (an app's |
| 81 | // re-read fails after the index moved, and the next index read is a 304) and |
| 82 | // codex 1 (a pass captures a snapshot and stalls while a later pass consumes |
| 83 | // a newer one). A failing seed reruns alone with -seed N. |
| 84 | |
| 85 | var ( |
| 86 | simSeedFlag = flag.Uint64("seed", 0, "TestRecordViewSimulation: run only this seed (0 runs the sweep)") |
| 87 | simSeedsFlag = flag.Int("seeds", 50, "TestRecordViewSimulation: how many seeds the sweep runs") |
| 88 | ) |
| 89 | |
| 90 | // simConsumeHandedOver is the consumer's mutation switch. The moves pass |
| 91 | // refreshes and takes the current snapshot inside w.pass and never acts on a |
| 92 | // snapshot handed over from the Reconcile leg (§4.3, codex 1). true makes the |
| 93 | // simulated consumer act on the handed-over one; a mutation check flips it. |
| 94 | var simConsumeHandedOver = false |
| 95 | |
| 96 | var simEpoch = time.Date(2026, 10, 21, 9, 0, 0, 0, time.UTC) |
| 97 | |
| 98 | var ( |
| 99 | simApps = []string{"shop", "blog", "docs", "wiki", "store", "fresh"} |
| 100 | simSlugs = []string{"one", "two", "three"} |
| 101 | simSidecars = []string{"bot", "cron"} |
| 102 | ) |
| 103 | |
| 104 | var errSimRead = errors.New("sim: read refused") |
| 105 | |
| 106 | // simParallel is how many seeds run at once. |
| 107 | const simParallel = 8 |
| 108 | |
| 109 | // simTemplate is the fixture's bucket, read once and copied into a fresh |
| 110 | // root per seed (copying files is far cheaper than writing them through a |
| 111 | // store that fsyncs and bumps). |
| 112 | type simTemplate struct { |
| 113 | files map[string][]byte |
| 114 | acct [3]string |
| 115 | } |
| 116 | |
| 117 | func newSimTemplate(t *testing.T) *simTemplate { |
| 118 | t.Helper() |
| 119 | f := newViewFixture(t, false, true) |
| 120 | keys, err := f.bypass.List(ctx, "") |
| 121 | if err != nil { |
| 122 | t.Fatal(err) |
| 123 | } |
| 124 | tmpl := &simTemplate{files: map[string][]byte{}, acct: f.acct} |
| 125 | for _, k := range keys { |
| 126 | body, _, err := f.bypass.Get(ctx, k) |
| 127 | if err != nil { |
| 128 | t.Fatal(err) |
| 129 | } |
| 130 | tmpl.files[k] = body |
| 131 | } |
| 132 | if _, ok := tmpl.files[release.ChangeIndexKey]; !ok { |
| 133 | t.Fatal("fixture: no change index") |
| 134 | } |
| 135 | return tmpl |
| 136 | } |
| 137 | |
| 138 | // simOp is one view operation (a Refresh or a Rebuild) as the store sees it. |
| 139 | type simOp struct { |
| 140 | kind string |
| 141 | nested bool |
| 142 | fail map[string]bool // keys and prefixes whose every read fails |
| 143 | triggers []*simTrigger |
| 144 | |
| 145 | // Filled in as it runs, under sim.mu. |
| 146 | polled bool |
| 147 | pollSeq int // writes completed when the op first read the index |
| 148 | walked bool |
| 149 | counts map[string]int |
| 150 | read map[string]bool // scopes it read |
| 151 | failed map[string]bool // scopes with a read that failed |
| 152 | failedAny bool |
| 153 | wrote bool // a trigger wrote during it |
| 154 | nestedRan bool // a concurrent Refresh ran inside it |
| 155 | err error |
| 156 | } |
| 157 | |
| 158 | func newSimOp(kind string) *simOp { |
| 159 | return &simOp{kind: kind, fail: map[string]bool{}, counts: map[string]int{}, read: map[string]bool{}, failed: map[string]bool{}} |
| 160 | } |
| 161 | |
| 162 | type simTrigger struct { |
| 163 | scope string |
| 164 | n int |
| 165 | what string |
| 166 | act func(op *simOp) |
| 167 | fired bool |
| 168 | } |
| 169 | |
| 170 | type simOpKey struct{} |
| 171 | |
| 172 | // simWrite is one planned write: decided by the seeded generator up front, |
| 173 | // so a trigger firing in a reader goroutine never touches the rng. |
| 174 | type simWrite struct { |
| 175 | w *simWriter |
| 176 | key string |
| 177 | del bool |
| 178 | soft bool // an account manifest written soft-deleted |
| 179 | } |
| 180 | |
| 181 | type simWriter struct { |
| 182 | name string |
| 183 | store *release.DirStore |
| 184 | failBump bool |
| 185 | } |
| 186 | |
| 187 | type simVersion struct { |
| 188 | seq int |
| 189 | body []byte // nil: absent |
| 190 | } |
| 191 | |
| 192 | type simHandover struct { |
| 193 | sn *ViewSnapshot |
| 194 | from int // step it was captured at |
| 195 | } |
| 196 | |
| 197 | // simStats counts what a sweep exercised, so a simulation that silently |
| 198 | // stopped exercising something fails instead of passing. |
| 199 | type simStats struct { |
| 200 | steps, writes, bumped, bypassed, failedBumps, paused int |
| 201 | refreshes, rebuilds, triggers, nested, deferred, sibling int |
| 202 | readFailures, checksB, checksE, consumed, incomplete int |
| 203 | } |
| 204 | |
| 205 | func (a *simStats) add(b simStats) { |
| 206 | a.steps += b.steps |
| 207 | a.writes += b.writes |
| 208 | a.bumped += b.bumped |
| 209 | a.bypassed += b.bypassed |
| 210 | a.failedBumps += b.failedBumps |
| 211 | a.paused += b.paused |
| 212 | a.refreshes += b.refreshes |
| 213 | a.rebuilds += b.rebuilds |
| 214 | a.triggers += b.triggers |
| 215 | a.nested += b.nested |
| 216 | a.deferred += b.deferred |
| 217 | a.sibling += b.sibling |
| 218 | a.readFailures += b.readFailures |
| 219 | a.checksB += b.checksB |
| 220 | a.checksE += b.checksE |
| 221 | a.consumed += b.consumed |
| 222 | a.incomplete += b.incomplete |
| 223 | } |
| 224 | |
| 225 | type sim struct { |
| 226 | t *testing.T |
| 227 | rng *rand.Rand |
| 228 | bucket bool |
| 229 | acct [3]string |
| 230 | acctID map[string]int64 |
| 231 | |
| 232 | root string |
| 233 | hold string // where holdIndex links the index |
| 234 | live *simStore // what the view reads |
| 235 | clean *viewCountingStore // the bucket as a reader sees it, for (b) |
| 236 | raw *release.DirStore // plain reads for bookkeeping |
| 237 | f *viewFixture // for readWalk |
| 238 | watch *release.ChangeWatch |
| 239 | v *RecordView |
| 240 | bumpW *simWriter |
| 241 | bypW *simWriter |
| 242 | failW *simWriter |
| 243 | keys []string // every family key the simulation tracks |
| 244 | step int |
| 245 | writes bool // ops may plan writes (false while a write is in flight) |
| 246 | paused bool |
| 247 | |
| 248 | writeMu sync.Mutex // serializes writes, so a bump is attributable |
| 249 | |
| 250 | mu sync.Mutex // guards the fields below |
| 251 | seq int // writes started |
| 252 | done int // writes completed (writes are serialized: done is the last seq) |
| 253 | hist map[string][]simVersion |
| 254 | bumped map[string][]int // key -> seqs of writes whose bump took |
| 255 | dirty map[string]bool // scopes written since the last operation was planned |
| 256 | deferred int |
| 257 | errs []string |
| 258 | trace []string |
| 259 | stats simStats |
| 260 | |
| 261 | checkMu sync.Mutex |
| 262 | lastEp uint64 |
| 263 | lastVer map[string]int |
| 264 | carry map[string]bool // (e): scopes whose last read failed |
| 265 | pass sync.Mutex // the fake w.pass |
| 266 | consumed uint64 |
| 267 | hand []simHandover |
| 268 | } |
| 269 | |
| 270 | func newSim(t *testing.T, tmpl *simTemplate, seed uint64, bucket bool) *sim { |
| 271 | t.Helper() |
| 272 | root := t.TempDir() |
| 273 | for k, b := range tmpl.files { |
| 274 | p := filepath.Join(root, filepath.FromSlash(k)) |
| 275 | if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { |
| 276 | t.Fatal(err) |
| 277 | } |
| 278 | if err := os.WriteFile(p, b, 0o644); err != nil { |
| 279 | t.Fatal(err) |
| 280 | } |
| 281 | } |
| 282 | s := &sim{ |
| 283 | t: t, rng: rand.New(rand.NewPCG(seed, 0x336)), bucket: bucket, acct: tmpl.acct, root: root, |
| 284 | acctID: map[string]int64{}, writes: true, |
| 285 | hist: map[string][]simVersion{}, bumped: map[string][]int{}, dirty: map[string]bool{}, lastVer: map[string]int{}, carry: map[string]bool{}, |
| 286 | } |
| 287 | for i, a := range tmpl.acct { |
| 288 | s.acctID[a] = int64(301 + i) |
| 289 | } |
| 290 | s.hold = filepath.Join(t.TempDir(), "changes.json") |
| 291 | s.live = &simStore{DirStore: release.NewDirStore(root), bucket: bucket, sim: s} |
| 292 | s.clean = &viewCountingStore{DirStore: release.NewDirStore(root), bucket: bucket} |
| 293 | s.raw = release.NewDirStore(root) |
| 294 | s.f = &viewFixture{live: s.clean, bypass: s.raw, acct: tmpl.acct} |
| 295 | s.watch = release.NewChangeWatch(s.live) |
| 296 | s.v = NewRecordView(s.live, s.watch) |
| 297 | bw, fw := release.NewDirStore(root), release.NewDirStore(root) |
| 298 | bw.EnableChangeIndex() |
| 299 | fw.EnableChangeIndex() |
| 300 | s.bumpW = &simWriter{name: "bumping", store: bw} |
| 301 | s.failW = &simWriter{name: "failing-bump", store: fw, failBump: true} |
| 302 | s.bypW = &simWriter{name: "bypassing", store: release.NewDirStore(root)} |
| 303 | |
| 304 | tracked := map[string]bool{} |
| 305 | for _, a := range s.acct { |
| 306 | tracked[accountManifestKey(a)] = true |
| 307 | for _, b := range simApps { |
| 308 | for _, slug := range simSlugs { |
| 309 | tracked[RecordKey(a, b, slug+"."+b+".example.com")] = true |
| 310 | } |
| 311 | for _, n := range simSidecars { |
| 312 | tracked[SidecarRecordKey(a, b, n)] = true |
| 313 | } |
| 314 | tracked[ManifestKey(a, b)] = true |
| 315 | tracked[SidecarManifestKey(a, b)] = true |
| 316 | tracked[release.AppTombstoneKey(a, b)] = true |
| 317 | } |
| 318 | } |
| 319 | for k := range tmpl.files { |
| 320 | if _, ok := (&ViewSnapshot{}).lookup(k); ok { |
| 321 | tracked[k] = true |
| 322 | } |
| 323 | } |
| 324 | for k := range tracked { |
| 325 | s.keys = append(s.keys, k) |
| 326 | var body []byte |
| 327 | if b, ok := tmpl.files[k]; ok { |
| 328 | body = b |
| 329 | } |
| 330 | s.hist[k] = []simVersion{{seq: 0, body: body}} |
| 331 | } |
| 332 | sort.Strings(s.keys) |
| 333 | return s |
| 334 | } |
| 335 | |
| 336 | func (s *sim) fail(format string, args ...any) { |
| 337 | s.mu.Lock() |
| 338 | defer s.mu.Unlock() |
| 339 | s.errs = append(s.errs, fmt.Sprintf(format, args...)) |
| 340 | } |
| 341 | |
| 342 | func (s *sim) tracef(format string, args ...any) { |
| 343 | s.mu.Lock() |
| 344 | defer s.mu.Unlock() |
| 345 | s.trace = append(s.trace, fmt.Sprintf("[%d] ", s.step)+fmt.Sprintf(format, args...)) |
| 346 | } |
| 347 | |
| 348 | // stop fails the test with the invariant breaches and the trace so far. |
| 349 | func (s *sim) stop() { |
| 350 | s.t.Helper() |
| 351 | s.mu.Lock() |
| 352 | errs, trace := slices.Clone(s.errs), slices.Clone(s.trace) |
| 353 | s.mu.Unlock() |
| 354 | if len(errs) == 0 { |
| 355 | return |
| 356 | } |
| 357 | if len(trace) > 80 { |
| 358 | trace = trace[len(trace)-80:] |
| 359 | } |
| 360 | s.t.Fatalf("bucket=%v step %d:\n %s\ntrace (last %d):\n %s", s.bucket, s.step, |
| 361 | strings.Join(errs, "\n "), len(trace), strings.Join(trace, "\n ")) |
| 362 | } |
| 363 | |
| 364 | // simScope is the change scope a read of target (a key or a listed prefix) |
| 365 | // belongs to, or "" for the top-level listings and the index. |
| 366 | func simScope(target string) string { |
| 367 | if rest, ok := strings.CutPrefix(target, "accounts/"); ok { |
| 368 | if a, _, ok := strings.Cut(rest, "/"); ok && a != "" { |
| 369 | return "accounts/" + a |
| 370 | } |
| 371 | return "" |
| 372 | } |
| 373 | if sc, ok := release.ChangeScope(target); ok { |
| 374 | return sc |
| 375 | } |
| 376 | return "" |
| 377 | } |
| 378 | |
| 379 | // read runs before every read the view makes: it attributes the read to the |
| 380 | // operation in ctx, fires that operation's triggers for it, and fails it if |
| 381 | // the plan says so. Reads outside an operation (the checks) pass untouched. |
| 382 | func (s *sim) read(ctx context.Context, target string) error { |
| 383 | op, _ := ctx.Value(simOpKey{}).(*simOp) |
| 384 | if op == nil { |
| 385 | return nil |
| 386 | } |
| 387 | s.mu.Lock() |
| 388 | if target == release.ChangeIndexKey { |
| 389 | if !op.polled { |
| 390 | op.polled, op.pollSeq = true, s.done |
| 391 | } |
| 392 | fail := op.fail[target] |
| 393 | if fail { |
| 394 | op.failedAny = true |
| 395 | s.stats.readFailures++ |
| 396 | } |
| 397 | s.mu.Unlock() |
| 398 | if fail { |
| 399 | return errSimRead |
| 400 | } |
| 401 | return nil |
| 402 | } |
| 403 | if target == "accounts/" { |
| 404 | op.walked = true |
| 405 | } |
| 406 | scope := simScope(target) |
| 407 | var fire []*simTrigger |
| 408 | if scope != "" { |
| 409 | n := op.counts[scope] |
| 410 | op.counts[scope] = n + 1 |
| 411 | for _, tr := range op.triggers { |
| 412 | if !tr.fired && tr.scope == scope && tr.n == n { |
| 413 | tr.fired = true |
| 414 | fire = append(fire, tr) |
| 415 | } |
| 416 | } |
| 417 | } |
| 418 | s.mu.Unlock() |
| 419 | for _, tr := range fire { |
| 420 | s.tracef(" %s: before read %d of %s (%s): %s", op.kind, tr.n, scope, target, tr.what) |
| 421 | s.mu.Lock() |
| 422 | s.stats.triggers++ |
| 423 | s.mu.Unlock() |
| 424 | tr.act(op) |
| 425 | } |
| 426 | s.mu.Lock() |
| 427 | defer s.mu.Unlock() |
| 428 | fail := op.fail[target] |
| 429 | if scope != "" { |
| 430 | op.read[scope] = true |
| 431 | if fail { |
| 432 | op.failed[scope] = true |
| 433 | } |
| 434 | } |
| 435 | if fail { |
| 436 | op.failedAny = true |
| 437 | s.stats.readFailures++ |
| 438 | return errSimRead |
| 439 | } |
| 440 | return nil |
| 441 | } |
| 442 | |
| 443 | // simStore is the bucket as the view reads it. |
| 444 | type simStore struct { |
| 445 | *release.DirStore |
| 446 | bucket bool |
| 447 | sim *sim |
| 448 | } |
| 449 | |
| 450 | func (s *simStore) BucketShapedListDirs() bool { return s.bucket } |
| 451 | |
| 452 | func (s *simStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 453 | if err := s.sim.read(ctx, key); err != nil { |
| 454 | return nil, "", err |
| 455 | } |
| 456 | return s.DirStore.Get(ctx, key) |
| 457 | } |
| 458 | |
| 459 | func (s *simStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) { |
| 460 | if err := s.sim.read(ctx, key); err != nil { |
| 461 | return nil, "", false, err |
| 462 | } |
| 463 | return s.DirStore.GetIfChanged(ctx, key, etag) |
| 464 | } |
| 465 | |
| 466 | func (s *simStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 467 | if err := s.sim.read(ctx, prefix); err != nil { |
| 468 | return nil, err |
| 469 | } |
| 470 | return s.DirStore.List(ctx, prefix) |
| 471 | } |
| 472 | |
| 473 | func (s *simStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) { |
| 474 | if err := s.sim.read(ctx, prefix); err != nil { |
| 475 | return nil, err |
| 476 | } |
| 477 | return s.DirStore.ListETags(ctx, prefix) |
| 478 | } |
| 479 | |
| 480 | func (s *simStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 481 | if err := s.sim.read(ctx, prefix); err != nil { |
| 482 | return nil, err |
| 483 | } |
| 484 | return s.DirStore.ListDirs(ctx, prefix) |
| 485 | } |
| 486 | |
| 487 | func (s *simStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| 488 | if err := s.sim.read(ctx, prefix); err != nil { |
| 489 | return nil, err |
| 490 | } |
| 491 | return s.DirStore.ListMod(ctx, prefix) |
| 492 | } |
| 493 | |
| 494 | // ---- writers ---- |
| 495 | |
| 496 | func (s *sim) indexGen(scope string) uint64 { |
| 497 | idx, _, err := release.ReadChangeIndex(ctx, s.raw) |
| 498 | if err != nil { |
| 499 | s.fail("sim: reading the index: %v", err) |
| 500 | return 0 |
| 501 | } |
| 502 | return idx.Entries[scope] |
| 503 | } |
| 504 | |
| 505 | func (s *sim) body(w simWrite, seq int) []byte { |
| 506 | var v any |
| 507 | if a, ok := domains.AccountManifestSqid(w.key); ok { |
| 508 | m := domains.AccountManifest{ID: s.acctID[a], Sqid: a, Name: fmt.Sprintf("Account v%d", seq), CreatedAt: simEpoch} |
| 509 | if w.soft { |
| 510 | m.DeletedAt = simEpoch.Add(time.Hour) |
| 511 | } |
| 512 | v = m |
| 513 | } else { |
| 514 | a, b, rest, _ := appKey(w.key) |
| 515 | switch { |
| 516 | case rest == "instances.json": |
| 517 | v = Manifest{Schema: 1, App: b, Domain: fmt.Sprintf("v%d.%s.example.com", seq, b)} |
| 518 | case rest == "sidecars.json": |
| 519 | v = SidecarManifest{Schema: 1, App: b, Args: []string{"run"}, Channel: fmt.Sprintf("c%d", seq)} |
| 520 | case rest == "deleted.json": |
| 521 | v = release.AppTombstone{Schema: 1, Account: a, App: b, DeletedAt: simEpoch, Actor: fmt.Sprintf("t%d", seq)} |
| 522 | case strings.HasPrefix(rest, "instances/"): |
| 523 | host := strings.TrimSuffix(strings.TrimPrefix(rest, "instances/"), ".json") |
| 524 | slug, _, _ := strings.Cut(host, ".") |
| 525 | v = Record{Schema: 1, Host: host, Slug: slug, App: b, Account: a, Region: fmt.Sprintf("r%d", seq), Channel: "stable", CreatedAt: simEpoch} |
| 526 | case strings.HasPrefix(rest, "sidecars/"): |
| 527 | name := strings.TrimSuffix(strings.TrimPrefix(rest, "sidecars/"), ".json") |
| 528 | v = SidecarRecord{Schema: 1, Name: name, App: b, Account: a, Host: SidecarHost(name, b, a), Channel: fmt.Sprintf("c%d", seq)} |
| 529 | default: |
| 530 | panic("sim: no body for " + w.key) |
| 531 | } |
| 532 | } |
| 533 | body, err := json.Marshal(v) |
| 534 | if err != nil { |
| 535 | panic(err) |
| 536 | } |
| 537 | return body |
| 538 | } |
| 539 | |
| 540 | // begin registers a write before it touches the bucket, so a reader that |
| 541 | // sees it can always name it. Caller holds writeMu. |
| 542 | func (s *sim) begin(w simWrite) (seq int, body []byte, etag string, before uint64) { |
| 543 | scope, _ := release.ChangeScope(w.key) |
| 544 | before = s.indexGen(scope) |
| 545 | if _, e, err := s.raw.Get(ctx, w.key); err == nil { |
| 546 | etag = e |
| 547 | } |
| 548 | s.mu.Lock() |
| 549 | s.seq++ |
| 550 | seq = s.seq |
| 551 | s.mu.Unlock() |
| 552 | if !w.del { |
| 553 | body = s.body(w, seq) |
| 554 | } |
| 555 | s.mu.Lock() |
| 556 | s.hist[w.key] = append(s.hist[w.key], simVersion{seq: seq, body: body}) |
| 557 | s.mu.Unlock() |
| 558 | return seq, body, etag, before |
| 559 | } |
| 560 | |
| 561 | // holdIndex gives the index file a second link, so the failing-bump |
| 562 | // writer's bump cannot read it (safefile.OpenFile refuses a file with more |
| 563 | // than one link). Caller holds writeMu. |
| 564 | func (s *sim) holdIndex() { |
| 565 | if err := os.Link(filepath.Join(s.root, filepath.FromSlash(release.ChangeIndexKey)), s.hold); err != nil { |
| 566 | s.fail("sim: holding the index: %v", err) |
| 567 | } |
| 568 | } |
| 569 | |
| 570 | // end classifies the write by what the index did, and completes it. |
| 571 | func (s *sim) end(w simWrite, seq int, before uint64, err error) { |
| 572 | if w.w.failBump { |
| 573 | if err := os.Remove(s.hold); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 574 | s.fail("sim: releasing the index: %v", err) |
| 575 | } |
| 576 | } |
| 577 | if err != nil { |
| 578 | s.fail("sim: %s write of %s: %v", w.w.name, w.key, err) |
| 579 | } |
| 580 | scope, _ := release.ChangeScope(w.key) |
| 581 | after := s.indexGen(scope) |
| 582 | s.mu.Lock() |
| 583 | defer s.mu.Unlock() |
| 584 | s.done = seq |
| 585 | s.stats.writes++ |
| 586 | s.dirty[scope] = true |
| 587 | class := "no bump" |
| 588 | switch { |
| 589 | case after == before+1: |
| 590 | s.bumped[w.key] = append(s.bumped[w.key], seq) |
| 591 | s.stats.bumped++ |
| 592 | class = fmt.Sprintf("bumped %s to %d", scope, after) |
| 593 | case after != before: |
| 594 | s.errs = append(s.errs, fmt.Sprintf("sim: write %d moved %s from %d to %d", seq, scope, before, after)) |
| 595 | case w.w.failBump: |
| 596 | s.stats.failedBumps++ |
| 597 | default: |
| 598 | s.stats.bypassed++ |
| 599 | } |
| 600 | op := "put" |
| 601 | if w.del { |
| 602 | op = "delete" |
| 603 | } |
| 604 | s.trace = append(s.trace, fmt.Sprintf("[%d] write %d: %s %s %s (%s)", s.step, seq, w.w.name, op, w.key, class)) |
| 605 | } |
| 606 | |
| 607 | func (s *sim) write(w simWrite) { |
| 608 | s.writeMu.Lock() |
| 609 | defer s.writeMu.Unlock() |
| 610 | seq, body, etag, before := s.begin(w) |
| 611 | if w.w.failBump { |
| 612 | s.holdIndex() |
| 613 | } |
| 614 | var err error |
| 615 | if w.del { |
| 616 | err = w.w.store.Delete(ctx, w.key) |
| 617 | } else { |
| 618 | _, err = w.w.store.Put(ctx, w.key, body, etag) |
| 619 | } |
| 620 | s.end(w, seq, before, err) |
| 621 | } |
| 622 | |
| 623 | // simGate is a request body that holds the write at its first Read until |
| 624 | // released: the write has begun and is not yet in the bucket. |
| 625 | type simGate struct { |
| 626 | r *bytes.Reader |
| 627 | started chan struct{} |
| 628 | release chan struct{} |
| 629 | once sync.Once |
| 630 | } |
| 631 | |
| 632 | func (g *simGate) Read(p []byte) (int, error) { |
| 633 | g.once.Do(func() { |
| 634 | close(g.started) |
| 635 | <-g.release |
| 636 | }) |
| 637 | return g.r.Read(p) |
| 638 | } |
| 639 | |
| 640 | func (g *simGate) Seek(off int64, whence int) (int64, error) { return g.r.Seek(off, whence) } |
| 641 | |
| 642 | // pausedWrite streams a put whose body stalls, runs view operations while |
| 643 | // it is in flight, then lets it finish and runs the next pass. |
| 644 | func (s *sim) pausedWrite(w simWrite, during func()) { |
| 645 | s.writeMu.Lock() |
| 646 | seq, body, etag, before := s.begin(w) |
| 647 | s.tracef(" write %d in flight: %s put %s", seq, w.w.name, w.key) |
| 648 | g := &simGate{r: bytes.NewReader(body), started: make(chan struct{}), release: make(chan struct{})} |
| 649 | done := make(chan error, 1) |
| 650 | go func() { |
| 651 | _, err := w.w.store.PutFrom(ctx, w.key, g, int64(len(body)), etag) |
| 652 | done <- err |
| 653 | }() |
| 654 | var err error |
| 655 | finished := false |
| 656 | select { |
| 657 | case <-g.started: |
| 658 | case err = <-done: |
| 659 | finished = true |
| 660 | } |
| 661 | if !finished { |
| 662 | s.paused, s.writes = true, false |
| 663 | s.mu.Lock() |
| 664 | s.stats.paused++ |
| 665 | s.mu.Unlock() |
| 666 | during() |
| 667 | s.paused, s.writes = false, true |
| 668 | if w.w.failBump { |
| 669 | s.holdIndex() // only now: the refreshes above read the index as usual |
| 670 | } |
| 671 | close(g.release) |
| 672 | err = <-done |
| 673 | } |
| 674 | s.end(w, seq, before, err) |
| 675 | s.writeMu.Unlock() |
| 676 | } |
| 677 | |
| 678 | // ---- view operations ---- |
| 679 | |
| 680 | // run is one view operation and the checks that follow every operation. |
| 681 | func (s *sim) run(op *simOp) error { |
| 682 | cctx := context.WithValue(context.Background(), simOpKey{}, op) |
| 683 | var err error |
| 684 | if op.kind == "refresh" { |
| 685 | err = s.v.Refresh(cctx) |
| 686 | } else { |
| 687 | err = s.v.Rebuild(cctx) |
| 688 | } |
| 689 | s.mu.Lock() |
| 690 | op.err = err |
| 691 | if op.kind == "refresh" { |
| 692 | s.stats.refreshes++ |
| 693 | } else { |
| 694 | s.stats.rebuilds++ |
| 695 | } |
| 696 | s.mu.Unlock() |
| 697 | sn := s.v.Snapshot() |
| 698 | var ep uint64 |
| 699 | complete := false |
| 700 | if sn != nil { |
| 701 | ep, complete = sn.Epoch, sn.Complete |
| 702 | } |
| 703 | var fails []string |
| 704 | for k := range op.fail { |
| 705 | fails = append(fails, k) |
| 706 | } |
| 707 | sort.Strings(fails) |
| 708 | 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) |
| 709 | if op.kind == "refresh" && err == nil && op.polled { |
| 710 | s.checkVisible(op) |
| 711 | } |
| 712 | s.checkSnapshot() |
| 713 | return err |
| 714 | } |
| 715 | |
| 716 | // top runs an operation as a pass would, then the (b) and (e) checks, then |
| 717 | // any concurrent Refresh that had to wait for mu. |
| 718 | func (s *sim) top(op *simOp) error { |
| 719 | carried := make(map[string]bool, len(s.carry)) |
| 720 | for sc := range s.carry { |
| 721 | carried[sc] = true |
| 722 | } |
| 723 | err := s.run(op) |
| 724 | |
| 725 | // (e): an incremental Refresh reads again every scope whose last read |
| 726 | // failed. A Refresh that walked read the whole deployment instead. |
| 727 | if op.kind == "refresh" && !op.walked { |
| 728 | for sc := range carried { |
| 729 | s.mu.Lock() |
| 730 | s.stats.checksE++ |
| 731 | s.mu.Unlock() |
| 732 | if !op.read[sc] { |
| 733 | s.fail("(e) %s failed its last read and this Refresh did not read it again", sc) |
| 734 | } |
| 735 | } |
| 736 | } |
| 737 | if op.walked && err == nil { |
| 738 | // A walk that got past its top-level listings read the whole |
| 739 | // deployment: a scope it did not reach has no keys left to read. |
| 740 | clear(s.carry) |
| 741 | } |
| 742 | for sc := range op.read { |
| 743 | if !op.failed[sc] { |
| 744 | delete(s.carry, sc) |
| 745 | } |
| 746 | } |
| 747 | for sc := range op.failed { |
| 748 | s.carry[sc] = true |
| 749 | } |
| 750 | |
| 751 | if op.kind == "rebuild" && err == nil && !op.failedAny && !op.wrote && !op.nestedRan && !s.paused { |
| 752 | s.checkMatchesLive() |
| 753 | } |
| 754 | |
| 755 | for { |
| 756 | s.mu.Lock() |
| 757 | n := s.deferred |
| 758 | s.deferred = 0 |
| 759 | s.mu.Unlock() |
| 760 | if n == 0 { |
| 761 | break |
| 762 | } |
| 763 | for range n { |
| 764 | s.mu.Lock() |
| 765 | s.stats.deferred++ |
| 766 | s.mu.Unlock() |
| 767 | s.top(newSimOp("refresh")) |
| 768 | } |
| 769 | } |
| 770 | return err |
| 771 | } |
| 772 | |
| 773 | // concurrent is a Refresh started by another caller at this instant. It gets |
| 774 | // in only if mu is free; otherwise it waits for the operation holding mu. |
| 775 | func (s *sim) concurrent(outer *simOp) { |
| 776 | if s.v.mu.TryLock() { |
| 777 | s.v.mu.Unlock() |
| 778 | s.mu.Lock() |
| 779 | outer.nestedRan = true |
| 780 | s.stats.nested++ |
| 781 | s.mu.Unlock() |
| 782 | op := newSimOp("refresh") |
| 783 | op.nested = true |
| 784 | s.run(op) |
| 785 | return |
| 786 | } |
| 787 | s.mu.Lock() |
| 788 | s.deferred++ |
| 789 | s.mu.Unlock() |
| 790 | } |
| 791 | |
| 792 | func (s *sim) siblingPoll() { |
| 793 | s.mu.Lock() |
| 794 | s.stats.sibling++ |
| 795 | s.mu.Unlock() |
| 796 | _, _ = s.watch.Poll(context.Background()) |
| 797 | } |
| 798 | |
| 799 | // ---- checks ---- |
| 800 | |
| 801 | // version is the newest write whose value the answer shows: a body names |
| 802 | // its write uniquely, and an absence is credited to the newest delete. |
| 803 | func (s *sim) version(key string, body []byte, absent bool) (int, bool) { |
| 804 | s.mu.Lock() |
| 805 | defer s.mu.Unlock() |
| 806 | best := -1 |
| 807 | for _, v := range s.hist[key] { |
| 808 | if (absent && v.body == nil) || (!absent && v.body != nil && bytes.Equal(v.body, body)) { |
| 809 | best = max(best, v.seq) |
| 810 | } |
| 811 | } |
| 812 | return best, best >= 0 |
| 813 | } |
| 814 | |
| 815 | func (s *sim) answer(st release.Store, key string) (body []byte, absent, failed bool) { |
| 816 | body, _, err := st.Get(context.Background(), key) |
| 817 | switch { |
| 818 | case err == nil: |
| 819 | return body, false, false |
| 820 | case errors.Is(err, os.ErrNotExist): |
| 821 | return nil, true, false |
| 822 | } |
| 823 | return nil, false, true |
| 824 | } |
| 825 | |
| 826 | // checkVisible is (a). |
| 827 | func (s *sim) checkVisible(op *simOp) { |
| 828 | sn := s.v.Snapshot() |
| 829 | if sn == nil { |
| 830 | s.fail("(a) a Refresh returned nil and published nothing") |
| 831 | return |
| 832 | } |
| 833 | st := sn.Store(s.live) |
| 834 | s.mu.Lock() |
| 835 | need := map[string]int{} |
| 836 | for key, seqs := range s.bumped { |
| 837 | for _, q := range seqs { |
| 838 | if q <= op.pollSeq { |
| 839 | need[key] = max(need[key], q) |
| 840 | } |
| 841 | } |
| 842 | } |
| 843 | s.mu.Unlock() |
| 844 | for key, q := range need { |
| 845 | body, absent, failed := s.answer(st, key) |
| 846 | if failed { |
| 847 | continue // a cached read failure: no consumer acts on it |
| 848 | } |
| 849 | ver, ok := s.version(key, body, absent) |
| 850 | switch { |
| 851 | case !ok: |
| 852 | s.fail("(a) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body) |
| 853 | case ver < q: |
| 854 | s.fail("(a) epoch %d: Refresh polled after write %d (bumped) completed, but %s shows write %d", sn.Epoch, q, key, ver) |
| 855 | } |
| 856 | } |
| 857 | } |
| 858 | |
| 859 | // checkSnapshot is (c) and (d) over the current snapshot, once per epoch. |
| 860 | func (s *sim) checkSnapshot() { |
| 861 | s.checkMu.Lock() |
| 862 | defer s.checkMu.Unlock() |
| 863 | sn := s.v.Snapshot() |
| 864 | if sn == nil { |
| 865 | return |
| 866 | } |
| 867 | if sn.Epoch < s.lastEp { |
| 868 | s.fail("(c) published epoch went from %d back to %d", s.lastEp, sn.Epoch) |
| 869 | return |
| 870 | } |
| 871 | if sn.Epoch == s.lastEp { |
| 872 | return |
| 873 | } |
| 874 | s.lastEp = sn.Epoch |
| 875 | if !sn.Complete { |
| 876 | s.mu.Lock() |
| 877 | s.stats.incomplete++ |
| 878 | s.mu.Unlock() |
| 879 | } |
| 880 | st := sn.Store(s.live) |
| 881 | bg := context.Background() |
| 882 | famErr := "" |
| 883 | for _, key := range s.keys { |
| 884 | body, absent, failed := s.answer(st, key) |
| 885 | if failed { |
| 886 | if famErr == "" { |
| 887 | famErr = "Get " + key |
| 888 | } |
| 889 | continue |
| 890 | } |
| 891 | ver, ok := s.version(key, body, absent) |
| 892 | if !ok { |
| 893 | s.fail("(c) epoch %d: %s holds a value no writer wrote: %q", sn.Epoch, key, body) |
| 894 | continue |
| 895 | } |
| 896 | if ver < s.lastVer[key] { |
| 897 | s.fail("(c) epoch %d: %s went back from write %d to write %d", sn.Epoch, key, s.lastVer[key], ver) |
| 898 | } |
| 899 | s.lastVer[key] = ver |
| 900 | } |
| 901 | for _, a := range s.acct { |
| 902 | if _, err := st.List(bg, "accounts/"+a+"/"); err != nil && famErr == "" { |
| 903 | famErr = "List accounts/" + a + "/" |
| 904 | } |
| 905 | if _, err := st.ListDirs(bg, "apps/"+a+"/"); err != nil && famErr == "" { |
| 906 | famErr = "ListDirs apps/" + a + "/" |
| 907 | } |
| 908 | for _, b := range simApps { |
| 909 | for _, p := range []string{RecordsPrefix(a, b), SidecarsPrefix(a, b)} { |
| 910 | if _, err := st.List(bg, p); err != nil && famErr == "" { |
| 911 | famErr = "List " + p |
| 912 | } |
| 913 | } |
| 914 | } |
| 915 | } |
| 916 | if _, rf, err := knownAccounts(bg, st); (err != nil || rf) && famErr == "" { |
| 917 | famErr = "knownAccounts" |
| 918 | } |
| 919 | _, _, strictErr := ListAllRecordsStrict(bg, st) |
| 920 | switch { |
| 921 | case sn.Complete && strictErr != nil: |
| 922 | s.fail("(d) epoch %d is Complete but ListAllRecordsStrict over it fails: %v", sn.Epoch, strictErr) |
| 923 | case sn.Complete && famErr != "": |
| 924 | s.fail("(d) epoch %d is Complete but %s over it fails", sn.Epoch, famErr) |
| 925 | case !sn.Complete && strictErr == nil && recordWalkGap(sn): |
| 926 | s.fail("(d) epoch %d is not Complete, its gap is in the record walk, and ListAllRecordsStrict over it returned nil", sn.Epoch) |
| 927 | case !sn.Complete && strictErr == nil && famErr == "": |
| 928 | s.fail("(d) epoch %d is not Complete, yet every reader over it answers as if it were", sn.Epoch) |
| 929 | } |
| 930 | } |
| 931 | |
| 932 | // recordWalkGap reports a gap in what ListAllRecordsStrict reads: the |
| 933 | // top-level walk, an account it would know (or fail to know), a known |
| 934 | // account's app directories, an instances listing or a record body. |
| 935 | func recordWalkGap(sn *ViewSnapshot) bool { |
| 936 | if sn.walkErr != nil { |
| 937 | return true |
| 938 | } |
| 939 | known, readFailed, err := knownAccounts(context.Background(), sn.Store(uncachedStore{})) |
| 940 | if err != nil || readFailed { |
| 941 | return true |
| 942 | } |
| 943 | for a := range known { |
| 944 | if sn.dirErrs[a] != nil { |
| 945 | return true |
| 946 | } |
| 947 | for _, e := range sn.apps[a] { |
| 948 | if e.instErr != nil { |
| 949 | return true |
| 950 | } |
| 951 | for _, ke := range e.inst { |
| 952 | if e.objs[ke.Key].failed() { |
| 953 | return true |
| 954 | } |
| 955 | } |
| 956 | } |
| 957 | } |
| 958 | return false |
| 959 | } |
| 960 | |
| 961 | // checkMatchesLive is (b): after a clean Rebuild with nothing written during |
| 962 | // it, the readers over the snapshot answer exactly as over the bucket. |
| 963 | func (s *sim) checkMatchesLive() { |
| 964 | s.stats.checksB++ |
| 965 | sn := s.v.Snapshot() |
| 966 | if !sn.Complete { |
| 967 | s.fail("(b) a clean Rebuild published epoch %d not Complete", sn.Epoch) |
| 968 | return |
| 969 | } |
| 970 | lr, lu, lerr := ListAllRecordsStrict(ctx, s.clean) |
| 971 | sr, su, serr := ListAllRecordsStrict(ctx, sn.Store(s.live)) |
| 972 | if lerr != nil || serr != nil { |
| 973 | s.fail("(b) ListAllRecordsStrict: bucket %v, snapshot %v", lerr, serr) |
| 974 | return |
| 975 | } |
| 976 | norm := func(rs []Record) []Record { |
| 977 | for i := range rs { |
| 978 | rs[i].CreatedAt = rs[i].CreatedAt.UTC() |
| 979 | } |
| 980 | return rs |
| 981 | } |
| 982 | if !reflect.DeepEqual(norm(lr), norm(sr)) || !reflect.DeepEqual(lu, su) { |
| 983 | s.fail("(b) ListAllRecordsStrict over epoch %d:\n%+v %v\nover the bucket:\n%+v %v", sn.Epoch, sr, su, lr, lu) |
| 984 | return |
| 985 | } |
| 986 | if want, got := readWalk(s.t, s.clean, s.f), readWalk(s.t, sn.Store(s.live), s.f); !reflect.DeepEqual(got, want) { |
| 987 | s.fail("(b) the record walk over epoch %d:\n%+v\nover the bucket:\n%+v", sn.Epoch, got, want) |
| 988 | } |
| 989 | } |
| 990 | |
| 991 | // ---- the consumer ---- |
| 992 | |
| 993 | // consumerBegin is a pass's Reconcile leg: refresh, take the snapshot, and |
| 994 | // hand it over towards w.pass. |
| 995 | func (s *sim) consumerBegin(op *simOp) { |
| 996 | s.top(op) |
| 997 | if sn := s.v.Snapshot(); sn != nil { |
| 998 | s.hand = append(s.hand, simHandover{sn: sn, from: s.step}) |
| 999 | s.tracef(" pass begins with epoch %d handed over", sn.Epoch) |
| 1000 | } |
| 1001 | } |
| 1002 | |
| 1003 | // consumerFinish is the moves pass reaching w.pass: it refreshes and takes |
| 1004 | // the current snapshot under it, records the epoch it consumed and refuses a |
| 1005 | // lower one. simConsumeHandedOver makes it use the handed-over snapshot. |
| 1006 | func (s *sim) consumerFinish(i int, op *simOp) { |
| 1007 | h := s.hand[i] |
| 1008 | s.hand = slices.Delete(s.hand, i, i+1) |
| 1009 | s.pass.Lock() |
| 1010 | defer s.pass.Unlock() |
| 1011 | sn, err := h.sn, error(nil) |
| 1012 | if !simConsumeHandedOver { |
| 1013 | err = s.top(op) |
| 1014 | sn = s.v.Snapshot() |
| 1015 | } |
| 1016 | if sn == nil { |
| 1017 | return |
| 1018 | } |
| 1019 | s.tracef(" moves pass (begun at step %d) consumes epoch %d (complete=%v, err=%v)", h.from, sn.Epoch, sn.Complete, err) |
| 1020 | s.stats.consumed++ |
| 1021 | if sn.Epoch < s.consumed { |
| 1022 | s.fail("(c) the moves pass consumed epoch %d after it had consumed %d", sn.Epoch, s.consumed) |
| 1023 | } |
| 1024 | s.consumed = max(s.consumed, sn.Epoch) |
| 1025 | if err == nil && sn.Complete { |
| 1026 | if _, _, err := ListAllRecordsStrict(context.Background(), sn.Store(s.live)); err != nil { |
| 1027 | s.fail("(d) the moves pass acted on Complete epoch %d and the strict reader failed: %v", sn.Epoch, err) |
| 1028 | } |
| 1029 | } |
| 1030 | } |
| 1031 | |
| 1032 | // ---- the generator ---- |
| 1033 | |
| 1034 | func (s *sim) pickWriter() *simWriter { |
| 1035 | switch r := s.rng.IntN(10); { |
| 1036 | case r < 6: |
| 1037 | return s.bumpW |
| 1038 | case r < 8: |
| 1039 | return s.bypW |
| 1040 | } |
| 1041 | return s.failW |
| 1042 | } |
| 1043 | |
| 1044 | func (s *sim) pickScope() string { |
| 1045 | a := s.acct[s.rng.IntN(len(s.acct))] |
| 1046 | if s.rng.IntN(100) < 15 { |
| 1047 | return "accounts/" + a |
| 1048 | } |
| 1049 | return "apps/" + a + "/" + simApps[s.rng.IntN(len(simApps))] |
| 1050 | } |
| 1051 | |
| 1052 | func (s *sim) pickWrite(scope string, w *simWriter, puts bool) simWrite { |
| 1053 | if a, ok := scopeAccount(scope); ok { |
| 1054 | return simWrite{w: w, key: accountManifestKey(a), soft: s.rng.IntN(100) < 15} |
| 1055 | } |
| 1056 | a, b, _ := ScopeApp(scope) |
| 1057 | slug := simSlugs[s.rng.IntN(len(simSlugs))] |
| 1058 | name := simSidecars[s.rng.IntN(len(simSidecars))] |
| 1059 | r := s.rng.IntN(20) |
| 1060 | if puts && r >= 8 && r < 11 { |
| 1061 | r = 0 |
| 1062 | } |
| 1063 | switch { |
| 1064 | case r < 8: |
| 1065 | return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com")} |
| 1066 | case r < 11: |
| 1067 | return simWrite{w: w, key: RecordKey(a, b, slug+"."+b+".example.com"), del: true} |
| 1068 | case r < 14: |
| 1069 | return simWrite{w: w, key: SidecarRecordKey(a, b, name), del: !puts && r == 13} |
| 1070 | case r < 16: |
| 1071 | return simWrite{w: w, key: ManifestKey(a, b)} |
| 1072 | case r < 17: |
| 1073 | return simWrite{w: w, key: SidecarManifestKey(a, b)} |
| 1074 | } |
| 1075 | return simWrite{w: w, key: release.AppTombstoneKey(a, b), del: !puts && r == 19} |
| 1076 | } |
| 1077 | |
| 1078 | func (s *sim) pickFailTarget() string { |
| 1079 | a := s.acct[s.rng.IntN(len(s.acct))] |
| 1080 | b := simApps[s.rng.IntN(len(simApps))] |
| 1081 | switch r := s.rng.IntN(100); { |
| 1082 | case r < 70: |
| 1083 | slug := simSlugs[s.rng.IntN(len(simSlugs))] |
| 1084 | return []string{ |
| 1085 | RecordsPrefix(a, b), SidecarsPrefix(a, b), RecordKey(a, b, slug+"."+b+".example.com"), |
| 1086 | SidecarRecordKey(a, b, simSidecars[s.rng.IntN(len(simSidecars))]), |
| 1087 | ManifestKey(a, b), SidecarManifestKey(a, b), release.AppTombstoneKey(a, b), |
| 1088 | }[s.rng.IntN(7)] |
| 1089 | case r < 85: |
| 1090 | return []string{"accounts/" + a + "/", accountManifestKey(a)}[s.rng.IntN(2)] |
| 1091 | } |
| 1092 | return []string{"accounts/", "apps/", "apps/" + a + "/", release.ChangeIndexKey}[s.rng.IntN(4)] |
| 1093 | } |
| 1094 | |
| 1095 | // triggerScope picks where a trigger waits. An incremental Refresh reads only |
| 1096 | // the scopes that moved or failed, so those are favoured; a Rebuild reads |
| 1097 | // every scope. |
| 1098 | func (s *sim) triggerScope(kind string) string { |
| 1099 | s.mu.Lock() |
| 1100 | var hot []string |
| 1101 | for sc := range s.dirty { |
| 1102 | hot = append(hot, sc) |
| 1103 | } |
| 1104 | s.mu.Unlock() |
| 1105 | for sc := range s.carry { |
| 1106 | if !slices.Contains(hot, sc) { |
| 1107 | hot = append(hot, sc) |
| 1108 | } |
| 1109 | } |
| 1110 | sort.Strings(hot) |
| 1111 | if kind == "refresh" && len(hot) > 0 && s.rng.IntN(100) < 70 { |
| 1112 | return hot[s.rng.IntN(len(hot))] |
| 1113 | } |
| 1114 | return s.pickScope() |
| 1115 | } |
| 1116 | |
| 1117 | // plan draws an operation: read failures, and triggers that write into the |
| 1118 | // scope being read, start a concurrent Refresh, or poll the shared watch. |
| 1119 | func (s *sim) plan(kind string) *simOp { |
| 1120 | op := newSimOp(kind) |
| 1121 | if s.rng.IntN(100) < 30 { |
| 1122 | for range 1 + s.rng.IntN(2) { |
| 1123 | op.fail[s.pickFailTarget()] = true |
| 1124 | } |
| 1125 | } |
| 1126 | defer func() { |
| 1127 | s.mu.Lock() |
| 1128 | clear(s.dirty) |
| 1129 | s.mu.Unlock() |
| 1130 | }() |
| 1131 | for range s.rng.IntN(4) { |
| 1132 | tr := &simTrigger{scope: s.triggerScope(kind), n: s.rng.IntN(6)} |
| 1133 | k := s.rng.IntN(20) |
| 1134 | if !s.writes && k < 15 { |
| 1135 | k = 15 + s.rng.IntN(5) |
| 1136 | } |
| 1137 | switch { |
| 1138 | case k < 12: |
| 1139 | wr := s.pickWrite(tr.scope, s.pickWriter(), false) |
| 1140 | tr.what = fmt.Sprintf("%s writes %s", wr.w.name, wr.key) |
| 1141 | tr.act = func(o *simOp) { s.hookWrite(o, wr) } |
| 1142 | case k < 15: |
| 1143 | // The codex-2 shape: a write lands behind the reader, and a |
| 1144 | // concurrent Refresh starts at once. |
| 1145 | wr := s.pickWrite(tr.scope, s.bumpW, true) |
| 1146 | tr.what = fmt.Sprintf("bumping writes %s, then a concurrent Refresh", wr.key) |
| 1147 | tr.act = func(o *simOp) { s.hookWrite(o, wr); s.concurrent(o) } |
| 1148 | case k < 18: |
| 1149 | tr.what = "a concurrent Refresh" |
| 1150 | tr.act = s.concurrent |
| 1151 | default: |
| 1152 | tr.what = "a sibling polls the shared watch" |
| 1153 | tr.act = func(*simOp) { s.siblingPoll() } |
| 1154 | } |
| 1155 | op.triggers = append(op.triggers, tr) |
| 1156 | } |
| 1157 | return op |
| 1158 | } |
| 1159 | |
| 1160 | func (s *sim) hookWrite(o *simOp, wr simWrite) { |
| 1161 | s.mu.Lock() |
| 1162 | o.wrote = true |
| 1163 | s.mu.Unlock() |
| 1164 | s.write(wr) |
| 1165 | } |
| 1166 | |
| 1167 | // viewStep draws one view-side step: a Refresh, a Rebuild, or one leg of a |
| 1168 | // moves pass. |
| 1169 | func (s *sim) viewStep() { |
| 1170 | switch r := s.rng.IntN(100); { |
| 1171 | case r < 45: |
| 1172 | s.tracef("refresh") |
| 1173 | s.top(s.plan("refresh")) |
| 1174 | case r < 60: |
| 1175 | s.tracef("rebuild") |
| 1176 | s.top(s.plan("rebuild")) |
| 1177 | case r < 75: |
| 1178 | s.tracef("pass begins") |
| 1179 | s.consumerBegin(s.plan("refresh")) |
| 1180 | case r < 90 && len(s.hand) > 0: |
| 1181 | s.tracef("moves pass reaches w.pass") |
| 1182 | s.consumerFinish(s.rng.IntN(len(s.hand)), s.plan("refresh")) |
| 1183 | default: |
| 1184 | s.tracef("quiet refresh") |
| 1185 | s.top(newSimOp("refresh")) |
| 1186 | } |
| 1187 | } |
| 1188 | |
| 1189 | func (s *sim) randomStep() { |
| 1190 | switch r := s.rng.IntN(100); { |
| 1191 | case r < 22: |
| 1192 | wr := s.pickWrite(s.pickScope(), s.pickWriter(), false) |
| 1193 | s.tracef("write") |
| 1194 | s.write(wr) |
| 1195 | case r < 30: |
| 1196 | w := s.bumpW |
| 1197 | if s.rng.IntN(10) < 3 { |
| 1198 | w = s.pickWriter() |
| 1199 | } |
| 1200 | wr := s.pickWrite(s.pickScope(), w, true) |
| 1201 | s.tracef("write in flight") |
| 1202 | s.pausedWrite(wr, func() { |
| 1203 | s.tracef("refresh while a write is in flight") |
| 1204 | s.top(s.plan("refresh")) |
| 1205 | if s.rng.IntN(2) == 0 { |
| 1206 | s.viewStep() |
| 1207 | } |
| 1208 | }) |
| 1209 | s.tracef("refresh after the write completed") |
| 1210 | s.top(s.plan("refresh")) |
| 1211 | case r < 33: |
| 1212 | s.tracef("sibling poll") |
| 1213 | s.siblingPoll() |
| 1214 | default: |
| 1215 | s.viewStep() |
| 1216 | } |
| 1217 | } |
| 1218 | |
| 1219 | func runSimSeed(t *testing.T, tmpl *simTemplate, seed uint64) simStats { |
| 1220 | s := newSim(t, tmpl, seed, seed%2 == 0) |
| 1221 | steps := 20 + s.rng.IntN(15) |
| 1222 | for s.step = 1; s.step <= steps; s.step++ { |
| 1223 | s.randomStep() |
| 1224 | s.stats.steps++ |
| 1225 | s.stop() |
| 1226 | } |
| 1227 | // Quiesce: a clean backstop, which (b) checks, then a clean refresh. |
| 1228 | s.tracef("final rebuild") |
| 1229 | s.top(newSimOp("rebuild")) |
| 1230 | s.tracef("final refresh") |
| 1231 | s.top(newSimOp("refresh")) |
| 1232 | s.stop() |
| 1233 | return s.stats |
| 1234 | } |
| 1235 | |
| 1236 | // ---- the named interleavings (§8) ---- |
| 1237 | |
| 1238 | func simScripted(t *testing.T, tmpl *simTemplate) { |
| 1239 | for _, bucket := range []bool{true, false} { |
| 1240 | // codex 2: the backstop lists an app before a record appears, and an |
| 1241 | // incremental refresh sees the record. The refresh must wait for the |
| 1242 | // backstop and publish after it, so the record is never published |
| 1243 | // and then lost. |
| 1244 | t.Run(fmt.Sprintf("codex2-backstop-vs-refresh/bucket=%v", bucket), func(t *testing.T) { |
| 1245 | s := newSim(t, tmpl, 1, bucket) |
| 1246 | s.top(newSimOp("rebuild")) |
| 1247 | a := s.acct[0] |
| 1248 | key := RecordKey(a, "shop", "three.shop.example.com") |
| 1249 | op := newSimOp("rebuild") |
| 1250 | op.triggers = []*simTrigger{{scope: "apps/" + a + "/shop", n: 1, what: "record appears, concurrent Refresh", |
| 1251 | act: func(o *simOp) { s.hookWrite(o, simWrite{w: s.bumpW, key: key}); s.concurrent(o) }}} |
| 1252 | s.top(op) |
| 1253 | s.stop() |
| 1254 | if !op.triggers[0].fired { |
| 1255 | t.Fatal("the trigger never fired") |
| 1256 | } |
| 1257 | if body, absent, failed := s.answer(s.v.Snapshot().Store(s.live), key); absent || failed || body == nil { |
| 1258 | t.Fatalf("the record is missing after the backstop and the refresh behind it (absent %v, failed %v)", absent, failed) |
| 1259 | } |
| 1260 | }) |
| 1261 | // codex 4: an app's re-read fails after the index moved, and the |
| 1262 | // next index read is a 304. The next Refresh must read it again. |
| 1263 | t.Run(fmt.Sprintf("codex4-failed-reread-then-304/bucket=%v", bucket), func(t *testing.T) { |
| 1264 | s := newSim(t, tmpl, 1, bucket) |
| 1265 | s.top(newSimOp("rebuild")) |
| 1266 | a := s.acct[0] |
| 1267 | key := RecordKey(a, "shop", "three.shop.example.com") |
| 1268 | s.write(simWrite{w: s.bumpW, key: key}) |
| 1269 | op := newSimOp("refresh") |
| 1270 | op.fail[RecordsPrefix(a, "shop")] = true |
| 1271 | s.top(op) |
| 1272 | if !op.failed["apps/"+a+"/shop"] { |
| 1273 | t.Fatal("the re-read did not fail") |
| 1274 | } |
| 1275 | _, before, _ := s.raw.Get(ctx, release.ChangeIndexKey) |
| 1276 | again := newSimOp("refresh") |
| 1277 | s.top(again) |
| 1278 | s.stop() |
| 1279 | if _, after, _ := s.raw.Get(ctx, release.ChangeIndexKey); after != before { |
| 1280 | t.Fatal("the index moved: the second poll was not a 304") |
| 1281 | } |
| 1282 | if !again.read["apps/"+a+"/shop"] { |
| 1283 | t.Fatal("the failed scope was not read again") |
| 1284 | } |
| 1285 | sn := s.v.Snapshot() |
| 1286 | if body, absent, failed := s.answer(sn.Store(s.live), key); absent || failed || body == nil || !sn.Complete { |
| 1287 | t.Fatalf("after the retry: absent %v, failed %v, complete %v", absent, failed, sn.Complete) |
| 1288 | } |
| 1289 | }) |
| 1290 | // codex 1: a pass captures a snapshot, stalls; another pass consumes |
| 1291 | // a newer one; the first then reaches w.pass. It must not consume an |
| 1292 | // older epoch than the second did. |
| 1293 | t.Run(fmt.Sprintf("codex1-stalled-pass/bucket=%v", bucket), func(t *testing.T) { |
| 1294 | s := newSim(t, tmpl, 1, bucket) |
| 1295 | s.top(newSimOp("rebuild")) |
| 1296 | s.consumerBegin(newSimOp("refresh")) |
| 1297 | s.write(simWrite{w: s.bumpW, key: RecordKey(s.acct[1], "docs", "three.docs.example.com")}) |
| 1298 | s.consumerBegin(newSimOp("refresh")) |
| 1299 | if s.hand[1].sn.Epoch <= s.hand[0].sn.Epoch { |
| 1300 | t.Fatalf("fixture: handovers at epochs %d and %d", s.hand[0].sn.Epoch, s.hand[1].sn.Epoch) |
| 1301 | } |
| 1302 | s.consumerFinish(1, newSimOp("refresh")) |
| 1303 | s.consumerFinish(0, newSimOp("refresh")) |
| 1304 | s.stop() |
| 1305 | }) |
| 1306 | } |
| 1307 | } |
| 1308 | |
| 1309 | func TestRecordViewSimulation(t *testing.T) { |
| 1310 | if testing.Short() { |
| 1311 | t.Skip("the invariant simulation is not run under -short") |
| 1312 | } |
| 1313 | // The view and the bump log every failed read and refused bump; the |
| 1314 | // simulation makes thousands on purpose. Serial: it sets the log output. |
| 1315 | out := log.Writer() |
| 1316 | log.SetOutput(io.Discard) |
| 1317 | t.Cleanup(func() { log.SetOutput(out) }) |
| 1318 | tmpl := newSimTemplate(t) |
| 1319 | |
| 1320 | t.Run("scripted", func(t *testing.T) { simScripted(t, tmpl) }) |
| 1321 | |
| 1322 | first, n := uint64(1), *simSeedsFlag |
| 1323 | if *simSeedFlag != 0 { |
| 1324 | first, n = *simSeedFlag, 1 |
| 1325 | } |
| 1326 | var ( |
| 1327 | mu sync.Mutex |
| 1328 | total simStats |
| 1329 | ran int |
| 1330 | failSeed uint64 |
| 1331 | next atomic.Uint64 |
| 1332 | stop atomic.Bool |
| 1333 | ) |
| 1334 | next.Store(first) |
| 1335 | start := time.Now() |
| 1336 | t.Run("sweep", func(t *testing.T) { |
| 1337 | // A seed spends most of its time waiting for DirStore's fsyncs, so a |
| 1338 | // few run at once. Each seed is its own bucket, view and writers; the |
| 1339 | // sweep stops starting seeds at the first failure. |
| 1340 | var wg sync.WaitGroup |
| 1341 | for range simParallel { |
| 1342 | wg.Go(func() { |
| 1343 | for !stop.Load() { |
| 1344 | seed := next.Add(1) - 1 |
| 1345 | if seed >= first+uint64(n) { |
| 1346 | return |
| 1347 | } |
| 1348 | var st simStats |
| 1349 | ok := t.Run(fmt.Sprintf("seed=%d", seed), func(t *testing.T) { st = runSimSeed(t, tmpl, seed) }) |
| 1350 | mu.Lock() |
| 1351 | if ok { |
| 1352 | total.add(st) |
| 1353 | ran++ |
| 1354 | } else if failSeed == 0 || seed < failSeed { |
| 1355 | failSeed = seed |
| 1356 | stop.Store(true) |
| 1357 | } |
| 1358 | mu.Unlock() |
| 1359 | } |
| 1360 | }) |
| 1361 | } |
| 1362 | wg.Wait() |
| 1363 | if failSeed != 0 { |
| 1364 | t.Logf("first failing seed: %d (rerun with -seed %d)", failSeed, failSeed) |
| 1365 | } |
| 1366 | }) |
| 1367 | t.Logf("%d seeds passed in %v: %+v", ran, time.Since(start).Round(time.Millisecond), total) |
| 1368 | if t.Failed() || ran < 50 { |
| 1369 | return |
| 1370 | } |
| 1371 | // A simulation that stopped exercising something would pass silently. |
| 1372 | exercised := map[string]int{ |
| 1373 | "bumped writes": total.bumped, "bypassing writes": total.bypassed, "failed bumps": total.failedBumps, |
| 1374 | "writes in flight": total.paused, "triggers": total.triggers, "concurrent refreshes that waited": total.deferred, |
| 1375 | "sibling polls": total.sibling, "read failures": total.readFailures, "(b) checks": total.checksB, |
| 1376 | "(e) checks": total.checksE, "consumed snapshots": total.consumed, "incomplete snapshots": total.incomplete, |
| 1377 | "rebuilds": total.rebuilds, |
| 1378 | } |
| 1379 | if runtime.GOOS != "linux" { |
| 1380 | // Only the Linux safefile refuses a file with two links, which is |
| 1381 | // how a bump is made to fail; elsewhere those writes bump. |
| 1382 | delete(exercised, "failed bumps") |
| 1383 | t.Logf("failed bumps are not exercised on %s", runtime.GOOS) |
| 1384 | } |
| 1385 | for name, c := range exercised { |
| 1386 | if c == 0 { |
| 1387 | t.Errorf("the sweep exercised no %s", name) |
| 1388 | } |
| 1389 | } |
| 1390 | } |
| 1391 | |
amadan