amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package provision
2
3import (
4 "context"
5 "errors"
6 "fmt"
7 "os"
8 "reflect"
9 "slices"
10 "sort"
11 "strings"
12 "sync"
13 "testing"
14 "time"
15
16 "github.com/carlosframework/platform/internal/domains"
17 "github.com/carlosframework/platform/internal/release"
18)
19
20// viewCountingStore is a DirStore that counts every read by method and key,
21// and can fail every listing of one prefix. Embedding the concrete store
22// forwards every optional extension; bucket decides whether it promises
23// bucket-shaped ListDirs (the S3 store does, DirStore does not).
24type viewCountingStore struct {
25 *release.DirStore
26 bucket bool
27
28 mu sync.Mutex
29 calls []string
30 failList string // List, ListETags and ListDirs of exactly this prefix fail
31 failGet string // Get and GetIfChanged of exactly this key fail
32 ghost string // ListETags lists this key, which is not there: deleted between list and read
33 cancelOn string // ListETags of exactly this prefix calls cancel first
34 cancel context.CancelFunc
35 stall string // GetIfChanged of exactly this key blocks until the caller's ctx ends
36}
37
38func (s *viewCountingStore) setStall(k string) { s.mu.Lock(); s.stall = k; s.mu.Unlock() }
39
40func (s *viewCountingStore) note(method, key string) {
41 s.mu.Lock()
42 s.calls = append(s.calls, method+" "+key)
43 s.mu.Unlock()
44}
45
46func (s *viewCountingStore) listFails(prefix string) bool {
47 s.mu.Lock()
48 defer s.mu.Unlock()
49 return s.failList != "" && prefix == s.failList
50}
51
52func (s *viewCountingStore) getFails(key string) bool {
53 s.mu.Lock()
54 defer s.mu.Unlock()
55 return s.failGet != "" && key == s.failGet
56}
57
58func (s *viewCountingStore) setFailList(p string) { s.mu.Lock(); s.failList = p; s.mu.Unlock() }
59func (s *viewCountingStore) setFailGet(k string) { s.mu.Lock(); s.failGet = k; s.mu.Unlock() }
60func (s *viewCountingStore) setGhost(k string) { s.mu.Lock(); s.ghost = k; s.mu.Unlock() }
61
62func (s *viewCountingStore) reset() {
63 s.mu.Lock()
64 s.calls = nil
65 s.mu.Unlock()
66}
67
68func (s *viewCountingStore) got() []string {
69 s.mu.Lock()
70 defer s.mu.Unlock()
71 out := slices.Clone(s.calls)
72 sort.Strings(out)
73 return out
74}
75
76func (s *viewCountingStore) BucketShapedListDirs() bool { return s.bucket }
77
78func (s *viewCountingStore) Get(ctx context.Context, key string) ([]byte, string, error) {
79 s.note("Get", key)
80 if s.getFails(key) {
81 return nil, "", errors.New("get refused")
82 }
83 return s.DirStore.Get(ctx, key)
84}
85
86func (s *viewCountingStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
87 s.note("GetIfChanged", key)
88 s.mu.Lock()
89 stall := s.stall != "" && key == s.stall
90 s.mu.Unlock()
91 if stall {
92 <-ctx.Done()
93 return nil, "", false, ctx.Err()
94 }
95 if s.getFails(key) {
96 return nil, "", false, errors.New("get refused")
97 }
98 return s.DirStore.GetIfChanged(ctx, key, etag)
99}
100
101func (s *viewCountingStore) List(ctx context.Context, prefix string) ([]string, error) {
102 s.note("List", prefix)
103 if err := ctx.Err(); err != nil {
104 return nil, err
105 }
106 if s.listFails(prefix) {
107 return nil, errors.New("list refused")
108 }
109 return s.DirStore.List(ctx, prefix)
110}
111
112func (s *viewCountingStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
113 s.note("ListETags", prefix)
114 s.mu.Lock()
115 if s.cancelOn != "" && prefix == s.cancelOn {
116 s.cancel()
117 }
118 s.mu.Unlock()
119 if err := ctx.Err(); err != nil {
120 return nil, err
121 }
122 if s.listFails(prefix) {
123 return nil, errors.New("list refused")
124 }
125 kes, err := s.DirStore.ListETags(ctx, prefix)
126 s.mu.Lock()
127 ghost := s.ghost
128 s.mu.Unlock()
129 if err == nil && ghost != "" && strings.HasPrefix(ghost, prefix) {
130 kes = append(kes, release.KeyETag{Key: ghost, ETag: "ghost"})
131 slices.SortFunc(kes, func(x, y release.KeyETag) int { return strings.Compare(x.Key, y.Key) })
132 }
133 return kes, err
134}
135
136func (s *viewCountingStore) ListDirs(ctx context.Context, prefix string) ([]string, error) {
137 s.note("ListDirs", prefix)
138 if s.listFails(prefix) {
139 return nil, errors.New("listdirs refused")
140 }
141 return s.DirStore.ListDirs(ctx, prefix)
142}
143
144func (s *viewCountingStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) {
145 s.note("ListMod", prefix)
146 return s.DirStore.ListMod(ctx, prefix)
147}
148
149// viewFixture is a deployment of three accounts and five apps: records,
150// sidecars, both manifests and a trash marker. live writes bump the change
151// index; bypass writes the same directory as an old binary would, without
152// bumping it.
153type viewFixture struct {
154 live *viewCountingStore
155 bypass *release.DirStore
156 acct [3]string
157}
158
159func newViewFixture(t *testing.T, bucket, indexed bool) *viewFixture {
160 t.Helper()
161 root := t.TempDir()
162 d := release.NewDirStore(root)
163 d.EnableChangeIndex()
164 f := &viewFixture{live: &viewCountingStore{DirStore: d, bucket: bucket}, bypass: release.NewDirStore(root)}
165 var w release.Store = f.live
166 if !indexed {
167 w = f.bypass
168 }
169 for i := range f.acct {
170 f.acct[i] = writeAccountManifest(t, w, int64(301+i), fmt.Sprintf("Account %d", i))
171 }
172 a, b, c := f.acct[0], f.acct[1], f.acct[2]
173 for _, m := range []struct{ acct, app string }{{a, "shop"}, {a, "blog"}, {b, "docs"}, {c, "store"}} {
174 if err := PutManifest(ctx, w, m.acct, m.app, Manifest{Schema: 1, App: m.app, Domain: m.app + ".example.com"}); err != nil {
175 t.Fatal(err)
176 }
177 }
178 f.record(t, w, a, "shop", "one")
179 f.record(t, w, a, "shop", "two")
180 f.record(t, w, a, "blog", "one")
181 f.record(t, w, b, "docs", "one")
182 f.record(t, w, c, "store", "one")
183 // A record that reads but does not decode: unreadable, by its path host.
184 if _, err := w.Put(ctx, RecordKey(c, "store", "bad.store.example.com"), []byte("{not json"), ""); err != nil {
185 t.Fatal(err)
186 }
187 for _, m := range []struct{ acct, app string }{{a, "shop"}, {b, "wiki"}} {
188 if err := PutSidecarManifest(ctx, w, m.acct, m.app, SidecarManifest{Schema: 1, App: m.app, Args: []string{"run"}, Channel: "stable"}); err != nil {
189 t.Fatal(err)
190 }
191 f.sidecar(t, w, m.acct, m.app, "bot")
192 }
193 f.trash(t, w, a, "blog")
194 // Keys outside the record families.
195 for _, k := range []string{
196 "apps/" + a + "/shop/logs/2026/10/21/chunk-1.jsonl",
197 "apps/" + a + "/shop/channels/stable.json",
198 "apps/" + a + "/shop/config.json",
199 "accounts/" + a + "/members/owner.json",
200 } {
201 if _, err := w.Put(ctx, k, []byte(`{}`), ""); err != nil {
202 t.Fatal(err)
203 }
204 }
205 f.live.reset()
206 return f
207}
208
209func viewRecord(acct, app, slug string) Record {
210 return Record{Schema: 1, Host: slug + "." + app + ".example.com", Slug: slug, App: app,
211 Account: acct, Region: "ie", Channel: "stable", CreatedAt: time.Now()}
212}
213
214func (f *viewFixture) record(t *testing.T, s release.Store, acct, app, slug string) {
215 t.Helper()
216 if err := CreateRecord(ctx, s, viewRecord(acct, app, slug)); err != nil {
217 t.Fatal(err)
218 }
219}
220
221func (f *viewFixture) sidecar(t *testing.T, s release.Store, acct, app, name string) {
222 t.Helper()
223 rec := SidecarRecord{Schema: 1, Name: name, App: app, Account: acct, Host: SidecarHost(name, app, acct), Channel: "stable"}
224 if err := CreateSidecarRecord(ctx, s, rec); err != nil {
225 t.Fatal(err)
226 }
227}
228
229func (f *viewFixture) trash(t *testing.T, s release.Store, acct, app string) {
230 t.Helper()
231 body := mustJSON(t, release.AppTombstone{Schema: 1, Account: acct, App: app, DeletedAt: time.Now().UTC(), Actor: "t"})
232 if _, err := s.Put(ctx, release.AppTombstoneKey(acct, app), body, ""); err != nil {
233 t.Fatal(err)
234 }
235}
236
237func (f *viewFixture) view() *RecordView {
238 return NewRecordView(f.live, release.NewChangeWatch(f.live))
239}
240
241func mustRebuild(t *testing.T, v *RecordView) *ViewSnapshot {
242 t.Helper()
243 if err := v.Rebuild(ctx); err != nil {
244 t.Fatal(err)
245 }
246 sn := v.Snapshot()
247 if sn == nil {
248 t.Fatal("Rebuild published nothing")
249 }
250 return sn
251}
252
253func mustRefresh(t *testing.T, v *RecordView) *ViewSnapshot {
254 t.Helper()
255 if err := v.Refresh(ctx); err != nil {
256 t.Fatal(err)
257 }
258 sn := v.Snapshot()
259 if sn == nil {
260 t.Fatal("Refresh published nothing")
261 }
262 return sn
263}
264
265// walkAnswer is everything the record consumers read, as one comparable
266// value: the strict record set and its unreadable hosts, the known accounts,
267// each app's sidecar records, both manifests and the trash marker.
268type walkAnswer struct {
269 Records []Record
270 Unreadable map[string]bool
271 Known map[string]bool
272 ReadFailed bool
273 PerApp map[string]string
274}
275
276func readWalk(t *testing.T, s release.Store, f *viewFixture) walkAnswer {
277 t.Helper()
278 recs, unreadable, err := ListAllRecordsStrict(ctx, s)
279 if err != nil {
280 t.Fatalf("ListAllRecordsStrict: %v", err)
281 }
282 for i := range recs {
283 recs[i].CreatedAt = recs[i].CreatedAt.UTC()
284 }
285 known, readFailed, err := knownAccounts(ctx, s)
286 if err != nil {
287 t.Fatalf("knownAccounts: %v", err)
288 }
289 per := map[string]string{}
290 for _, acct := range f.acct {
291 for _, app := range []string{"shop", "blog", "docs", "wiki", "store", "fresh"} {
292 m, merr := GetManifest(ctx, s, acct, app)
293 sm, serr := GetSidecarManifest(ctx, s, acct, app)
294 tomb, terr := release.ReadAppTombstone(ctx, s, acct, app)
295 sides, _, lerr := ListSidecarRecords(ctx, s, acct, app, 100)
296 per[acct+"/"+app] = fmt.Sprintf("m=%s/%v sm=%s/%v tomb=%v/%v sides=%s/%v",
297 mustJSON(t, m), errClass(merr), mustJSON(t, sm), errClass(serr), tomb != nil, errClass(terr), mustJSON(t, sides), errClass(lerr))
298 }
299 }
300 return walkAnswer{Records: recs, Unreadable: unreadable, Known: known, ReadFailed: readFailed, PerApp: per}
301}
302
303func errClass(err error) string {
304 switch {
305 case err == nil:
306 return "ok"
307 case errors.Is(err, os.ErrNotExist):
308 return "absent"
309 }
310 return "error: " + err.Error()
311}
312
313func assertMatchesWalk(t *testing.T, f *viewFixture, sn *ViewSnapshot) {
314 t.Helper()
315 want := readWalk(t, f.live, f)
316 got := readWalk(t, sn.Store(f.live), f)
317 if !reflect.DeepEqual(got, want) {
318 t.Fatalf("over the snapshot:\n%+v\nover the bucket:\n%+v", got, want)
319 }
320}
321
322func viewHosts(t *testing.T, s release.Store) []string {
323 t.Helper()
324 recs, err := ListAllRecords(ctx, s)
325 if err != nil {
326 t.Fatal(err)
327 }
328 var hosts []string
329 for _, r := range recs {
330 hosts = append(hosts, r.Host)
331 }
332 return hosts
333}
334
335func TestViewMatchesWalk(t *testing.T) {
336 t.Parallel()
337 for _, bucket := range []bool{false, true} {
338 t.Run(fmt.Sprintf("bucket=%v", bucket), func(t *testing.T) {
339 t.Parallel()
340 f := newViewFixture(t, bucket, true)
341 v := f.view()
342 sn := mustRefresh(t, v)
343 if !sn.Complete {
344 t.Fatal("a clean first build is not complete")
345 }
346 if !release.HasBucketListDirs(sn.Store(f.live)) {
347 t.Fatal("the snapshot store does not promise bucket-shaped ListDirs, so the record walk would list apps/ whole")
348 }
349 assertMatchesWalk(t, f, sn)
350 if len(readWalk(t, f.live, f).Records) != 5 {
351 t.Fatal("fixture: want five readable records")
352 }
353
354 // Incremental changes in every family, then a Refresh.
355 a, b, c := f.acct[0], f.acct[1], f.acct[2]
356 f.record(t, f.live, b, "docs", "two")
357 moved := viewRecord(a, "shop", "one")
358 moved.Region = "us"
359 putReplace(t, f.live, RecordKey(a, "shop", moved.Host), mustJSON(t, moved))
360 f.record(t, f.live, a, "fresh", "one")
361 f.trash(t, f.live, a, "shop")
362 if err := f.live.Delete(ctx, release.AppTombstoneKey(a, "blog")); err != nil {
363 t.Fatal(err)
364 }
365 if err := f.live.Delete(ctx, SidecarRecordKey(b, "wiki", "bot")); err != nil {
366 t.Fatal(err)
367 }
368 if err := PutManifest(ctx, f.live, c, "store", Manifest{Schema: 1, App: "store", Domain: "other.example.com"}); err != nil {
369 t.Fatal(err)
370 }
371 soft := domains.AccountManifest{ID: 303, Sqid: c, Name: "gone", CreatedAt: time.Now().UTC(), DeletedAt: time.Now().UTC()}
372 putReplace(t, f.live, "accounts/"+c+"/account.json", mustJSON(t, soft))
373 sn2 := mustRefresh(t, v)
374 if sn2.Epoch <= sn.Epoch || !sn2.Complete {
375 t.Fatalf("refresh published epoch %d complete %v after %d", sn2.Epoch, sn2.Complete, sn.Epoch)
376 }
377 assertMatchesWalk(t, f, sn2)
378 })
379 }
380}
381
382// Complete is false exactly when the walk over the bucket would be partial:
383// an account whose apps will not list, or an account manifest that will not
384// read cleanly. A strict reader over the snapshot fails as it does live.
385func TestViewCompleteMirrorsTheWalk(t *testing.T) {
386 t.Parallel()
387 f := newViewFixture(t, true, true)
388 a, b := f.acct[0], f.acct[1]
389 v := f.view()
390 f.live.setFailList("apps/" + b + "/")
391 f.live.setGhost(RecordKey(a, "shop", "ghost.shop.example.com"))
392 sn := mustRebuild(t, v)
393 if sn.Complete {
394 t.Fatal("a walk that could not list an account's apps is complete")
395 }
396 if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
397 t.Fatal("a strict reader over a partial walk did not fail")
398 }
399 f.live.setFailList("")
400 sn = mustRefresh(t, v)
401 if !sn.Complete {
402 t.Fatal("the refresh after a partial walk did not walk again")
403 }
404 assertMatchesWalk(t, f, sn)
405
406 putReplace(t, f.live, "accounts/"+b+"/account.json", []byte("{not json"))
407 sn = mustRefresh(t, v)
408 if sn.Complete {
409 t.Fatal("a snapshot holding a malformed account manifest is complete")
410 }
411 _, _, lerr := ListAllRecordsStrict(ctx, f.live)
412 _, _, serr := ListAllRecordsStrict(ctx, sn.Store(f.live))
413 if lerr == nil || serr == nil {
414 t.Fatalf("strict readers over a malformed manifest: live %v, snapshot %v; want both to fail", lerr, serr)
415 }
416
417 // An account listing that fails: its manifest reads as failed, not absent.
418 f.live.setFailList("accounts/" + b + "/")
419 putReplace(t, f.live, "accounts/"+b+"/account.json", accountManifestBody(302, b, "Account 1"))
420 sn = mustRefresh(t, v)
421 if sn.Complete {
422 t.Fatal("a snapshot whose account listing failed is complete")
423 }
424 if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
425 t.Fatal("a strict reader over a failed account listing did not fail")
426 }
427 f.live.setFailList("")
428 if sn = mustRefresh(t, v); !sn.Complete {
429 t.Fatal("still incomplete after the account listing recovered")
430 }
431 assertMatchesWalk(t, f, sn)
432}
433
434// A walk that cannot list one account's apps does not know them: Get and List
435// of their records and manifests over the snapshot answer that failure, never
436// "absent" or "empty", until a refresh or walk reads them. Found by the
437// invariant simulation (a snapshot answered absent for keys no one deleted).
438func TestViewUnlistedAccountIsNotAbsent(t *testing.T) {
439 t.Parallel()
440 f := newViewFixture(t, true, true)
441 b := f.acct[1]
442 f.live.setFailList("apps/" + b + "/")
443 st := mustRebuild(t, f.view()).Store(f.live)
444 for _, key := range []string{ManifestKey(b, "docs"), RecordKey(b, "docs", "one.docs.example.com"), SidecarRecordKey(b, "wiki", "bot")} {
445 if _, _, err := st.Get(ctx, key); err == nil || errors.Is(err, os.ErrNotExist) {
446 t.Fatalf("Get %s over a walk that could not list %s's apps = %v, want the listing failure", key, b, err)
447 }
448 }
449 for _, p := range []string{RecordsPrefix(b, "docs"), SidecarsPrefix(b, "wiki")} {
450 if keys, err := st.List(ctx, p); err == nil {
451 t.Fatalf("List %s over a walk that could not list %s's apps = %v, nil; want the listing failure", p, b, keys)
452 }
453 }
454 if _, err := GetManifest(ctx, st, f.acct[0], "shop"); err != nil {
455 t.Fatalf("an account whose apps listed: %v", err)
456 }
457}
458
459// A refresh whose caller goes away mid-read publishes what it read, with the
460// reads that died cached as failures: the view knows it is behind, so it must
461// not leave the previous snapshot standing as Complete.
462func TestViewCancelledRefreshFailsClosed(t *testing.T) {
463 t.Parallel()
464 f := newViewFixture(t, true, true)
465 v := f.view()
466 before := mustRebuild(t, v)
467 a := f.acct[0]
468 f.record(t, f.live.DirStore, a, "shop", "three")
469 cctx, cancel := context.WithCancel(ctx)
470 defer cancel()
471 f.live.mu.Lock()
472 f.live.cancelOn, f.live.cancel = RecordsPrefix(a, "shop"), cancel
473 f.live.mu.Unlock()
474 if err := v.Refresh(cctx); err == nil {
475 t.Fatal("a refresh cancelled mid-read reported success")
476 }
477 sn := v.Snapshot()
478 if sn == before || sn.Complete {
479 t.Fatalf("after a cancelled refresh: same snapshot %v, complete %v; want a new, incomplete one", sn == before, sn.Complete)
480 }
481 if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
482 t.Fatal("a strict reader over a refresh that died mid-read did not fail")
483 }
484 f.live.mu.Lock()
485 f.live.cancelOn = ""
486 f.live.mu.Unlock()
487 sn = mustRefresh(t, v)
488 if !sn.Complete {
489 t.Fatal("still incomplete after the next refresh")
490 }
491 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
492 t.Fatalf("records after the next refresh = %v", hosts)
493 }
494}
495
496// A walk that fails at the top (the index unusable, then accounts/ will not
497// list) must not leave the last good snapshot standing as Complete: the
498// unchanged readers over it fail as they fail on the bucket.
499func TestViewFailedWalkFailsClosed(t *testing.T) {
500 t.Parallel()
501 f := newViewFixture(t, true, true)
502 v := f.view()
503 good := mustRebuild(t, v)
504 f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
505 f.live.setFailGet(release.ChangeIndexKey)
506 f.live.setFailList("accounts/")
507 if err := v.Refresh(ctx); err == nil {
508 t.Fatal("a refresh whose walk failed reported success")
509 }
510 sn := v.Snapshot()
511 if sn == nil || sn.Epoch <= good.Epoch || sn.Complete {
512 t.Fatalf("after a failed walk: %+v; want a newer, incomplete snapshot", sn)
513 }
514 st := sn.Store(f.live)
515 if _, _, err := ListAllRecordsStrict(ctx, st); err == nil {
516 t.Fatal("a strict reader over a failed walk did not fail")
517 }
518 if _, _, err := knownAccounts(ctx, st); err == nil {
519 t.Fatal("knownAccounts over a failed walk did not fail")
520 }
521 if _, err := st.ListDirs(ctx, "apps/"); err == nil {
522 t.Fatal("ListDirs(apps/) over a failed walk did not fail")
523 }
524 f.live.setFailGet("")
525 f.live.setFailList("")
526 sn = mustRefresh(t, v)
527 if !sn.Complete {
528 t.Fatal("still incomplete after the walk recovered")
529 }
530 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
531 t.Fatalf("records after the recovered walk = %v", hosts)
532 }
533}
534
535// Over a live store whose ListDirs is not bucket-shaped (the fleet door's
536// ListDirs("apps/") is its own account's apps), a write through the snapshot
537// store must not send the walk's ListDirs to live: walked as accounts, those
538// app directories hold no records, and reconcile would retire every route.
539func TestSnapshotStoreWriteOverNonBucketLive(t *testing.T) {
540 t.Parallel()
541 f := newViewFixture(t, false, true)
542 a := f.acct[0]
543 live := fleetShaped{Store: f.live, acct: a}
544 v := NewRecordView(live, release.NewChangeWatch(live))
545 sn := mustRefresh(t, v)
546 st := sn.Store(live)
547 rec := viewRecord(a, "fresh", "one")
548 if err := CreateRecord(ctx, st, rec); err != nil {
549 t.Fatal(err)
550 }
551 want := viewHosts(t, f.live)
552 if !slices.Contains(want, rec.Host) || len(want) != 6 {
553 t.Fatalf("fixture: the bucket holds %v", want)
554 }
555 if got := viewHosts(t, st); !reflect.DeepEqual(got, want) {
556 t.Fatalf("records over the snapshot store after a write = %v, want %v", got, want)
557 }
558 if dirs, err := st.ListDirs(ctx, "apps/"+a+"/"); err != nil || !slices.Contains(dirs, "apps/"+a+"/fresh/") {
559 t.Fatalf("ListDirs after the write = %v, %v; want the new app", dirs, err)
560 }
561}
562
563func TestViewRefreshRereadsOnlyMovedApp(t *testing.T) {
564 t.Parallel()
565 f := newViewFixture(t, true, true)
566 v := f.view()
567 mustRebuild(t, v)
568 a := f.acct[0]
569 f.record(t, f.live.DirStore, a, "shop", "three")
570 f.live.reset()
571 sn := mustRefresh(t, v)
572 app := "apps/" + a + "/shop/"
573 want := []string{
574 "GetIfChanged " + app + "deleted.json",
575 "GetIfChanged " + app + "instances.json",
576 "GetIfChanged " + app + "instances/three.shop.example.com.json",
577 "GetIfChanged " + app + "sidecars.json",
578 "GetIfChanged " + release.ChangeIndexKey,
579 "ListETags " + app + "instances/",
580 "ListETags " + app + "sidecars/",
581 }
582 if got := f.live.got(); !reflect.DeepEqual(got, want) {
583 t.Fatalf("one record in shop read\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
584 }
585 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
586 t.Fatalf("records over the snapshot = %v, want the new one", hosts)
587 }
588}
589
590func TestViewRefreshNoChangeIsOneGet(t *testing.T) {
591 t.Parallel()
592 f := newViewFixture(t, true, true)
593 v := f.view()
594 mustRebuild(t, v)
595 f.live.reset()
596 mustRefresh(t, v)
597 if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
598 t.Fatalf("a quiet refresh read %v, want %v", got, want)
599 }
600}
601
602func TestViewRetriesFailedScopeOn304(t *testing.T) {
603 t.Parallel()
604 f := newViewFixture(t, true, true)
605 v := f.view()
606 mustRebuild(t, v)
607 a := f.acct[0]
608 f.record(t, f.live.DirStore, a, "shop", "three")
609 f.live.setFailList(RecordsPrefix(a, "shop"))
610 sn := mustRefresh(t, v)
611 if sn.Complete {
612 t.Fatal("a scope whose re-read failed is reported complete")
613 }
614 if _, _, err := ListAllRecordsStrict(ctx, sn.Store(f.live)); err == nil {
615 t.Fatal("a strict reader over an incomplete snapshot did not fail")
616 }
617 if _, complete, err := ListAllRecordsTolerant(ctx, sn.Store(f.live)); err != nil || complete {
618 t.Fatalf("a tolerant reader over an incomplete snapshot = complete %v, %v", complete, err)
619 }
620
621 f.live.setFailList("")
622 f.live.reset()
623 sn = mustRefresh(t, v)
624 if !slices.Contains(f.live.got(), "ListETags "+RecordsPrefix(a, "shop")) {
625 t.Fatalf("the failed scope was not re-read on an unchanged index: %v", f.live.got())
626 }
627 if slices.Contains(f.live.got(), "ListETags accounts/") || slices.Contains(f.live.got(), "List accounts/") {
628 t.Fatalf("the retry walked the deployment: %v", f.live.got())
629 }
630 if !sn.Complete {
631 t.Fatal("still incomplete after the retry succeeded")
632 }
633 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
634 t.Fatalf("records after the retry = %v", hosts)
635 }
636}
637
638// A stalled index GET never holds a refresh, or the backstop rebuild, for
639// longer than the poll bound, whatever the caller's context: each gives up
640// on the index, walks the bucket (which answers), and publishes. The watch
641// is unknown meanwhile, and the next good poll brings the view back to the
642// index.
643func TestViewStalledIndexIsBounded(t *testing.T) {
644 t.Parallel()
645 f := newViewFixture(t, true, true)
646 v := f.view()
647 v.pollTimeout = 50 * time.Millisecond
648 mustRebuild(t, v)
649 f.record(t, f.live.DirStore, f.acct[1], "docs", "two")
650 f.live.setStall(release.ChangeIndexKey)
651 f.live.reset()
652 for _, step := range []struct {
653 name string
654 run func(context.Context) error
655 }{{"Refresh", v.Refresh}, {"Rebuild", v.Rebuild}} {
656 done := make(chan error, 1)
657 go func() { done <- step.run(context.Background()) }()
658 select {
659 case err := <-done:
660 if err != nil {
661 t.Fatalf("%s over a stalled index: %v", step.name, err)
662 }
663 case <-time.After(10 * time.Second):
664 t.Fatalf("%s over a stalled index did not return: its poll is unbounded", step.name)
665 }
666 sn := v.Snapshot()
667 if !sn.Complete || !slices.Contains(viewHosts(t, sn.Store(f.live)), "two.docs.example.com") {
668 t.Fatalf("%s over a stalled index: complete %v, records %v; want the walk's", step.name, sn.Complete, viewHosts(t, sn.Store(f.live)))
669 }
670 if v.watch.Usable() {
671 t.Fatalf("%s: the watch vouches after its poll gave up", step.name)
672 }
673 }
674 if n := slices.Index(f.live.got(), "GetIfChanged "+release.ChangeIndexKey); n < 0 {
675 t.Fatalf("no index poll at all: %v", f.live.got())
676 }
677 polls := 0
678 for _, c := range f.live.got() {
679 if c == "GetIfChanged "+release.ChangeIndexKey {
680 polls++
681 }
682 }
683 if polls != 2 {
684 t.Fatalf("the index was polled %d time(s) for one Refresh and one Rebuild, want 2: a refresh that gave up polls again before its walk", polls)
685 }
686 f.live.setStall("")
687 f.live.reset()
688 mustRefresh(t, v)
689 if !v.watch.Usable() {
690 t.Fatal("the watch did not recover on the next good poll")
691 }
692 f.live.reset()
693 mustRefresh(t, v)
694 if got, want := f.live.got(), []string{"GetIfChanged " + release.ChangeIndexKey}; !reflect.DeepEqual(got, want) {
695 t.Fatalf("a quiet refresh after recovery read %v, want %v", got, want)
696 }
697}
698
699func TestViewFirstRefreshWithoutIndexRebuilds(t *testing.T) {
700 t.Parallel()
701 f := newViewFixture(t, true, false)
702 if _, _, err := f.bypass.Get(ctx, release.ChangeIndexKey); !errors.Is(err, os.ErrNotExist) {
703 t.Fatalf("fixture wrote an index: %v", err)
704 }
705 v := f.view()
706 sn := mustRefresh(t, v)
707 if !sn.Complete {
708 t.Fatal("a first build over a bucket with no index is not complete")
709 }
710 assertMatchesWalk(t, f, sn)
711 if len(viewHosts(t, sn.Store(f.live))) != 5 {
712 t.Fatal("the first build missed records")
713 }
714}
715
716func TestViewUnusableWatchRebuilds(t *testing.T) {
717 t.Parallel()
718 f := newViewFixture(t, true, true)
719 v := f.view()
720 mustRebuild(t, v)
721 putReplace(t, f.bypass, release.ChangeIndexKey, []byte("{not json"))
722 f.record(t, f.bypass, f.acct[1], "docs", "two")
723 f.live.reset()
724 sn := mustRefresh(t, v)
725 if !slices.Contains(f.live.got(), "ListETags accounts/") {
726 t.Fatalf("an unusable index did not walk: %v", f.live.got())
727 }
728 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "two.docs.example.com") {
729 t.Fatalf("records after the walk = %v", hosts)
730 }
731}
732
733func TestViewBypassWriteAppearsAfterRebuild(t *testing.T) {
734 t.Parallel()
735 f := newViewFixture(t, true, true)
736 v := f.view()
737 mustRebuild(t, v)
738 f.record(t, f.bypass, f.acct[0], "shop", "three")
739 sn := mustRefresh(t, v)
740 if hosts := viewHosts(t, sn.Store(f.live)); slices.Contains(hosts, "three.shop.example.com") {
741 t.Fatal("a write that bypassed the index appeared without a walk")
742 }
743 sn = mustRebuild(t, v)
744 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
745 t.Fatalf("records after the backstop walk = %v", hosts)
746 }
747 assertMatchesWalk(t, f, sn)
748}
749
750func TestViewPurgeDropsApp(t *testing.T) {
751 t.Parallel()
752 f := newViewFixture(t, true, true)
753 v := f.view()
754 sn := mustRebuild(t, v)
755 a := f.acct[0]
756 dirs, err := sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
757 if err != nil || !slices.Contains(dirs, "apps/"+a+"/shop/") {
758 t.Fatalf("ListDirs before the purge = %v, %v", dirs, err)
759 }
760 keys, err := f.live.DirStore.List(ctx, "apps/"+a+"/shop/")
761 if err != nil {
762 t.Fatal(err)
763 }
764 for _, k := range keys {
765 if err := f.live.DirStore.Delete(ctx, k); err != nil {
766 t.Fatal(err)
767 }
768 }
769 sn = mustRefresh(t, v)
770 dirs, err = sn.Store(f.live).ListDirs(ctx, "apps/"+a+"/")
771 if err != nil || slices.Contains(dirs, "apps/"+a+"/shop/") || !slices.Contains(dirs, "apps/"+a+"/blog/") {
772 t.Fatalf("ListDirs after the purge = %v, %v; want blog and no shop", dirs, err)
773 }
774 assertMatchesWalk(t, f, sn)
775}
776
777func TestViewEpochsIncrease(t *testing.T) {
778 t.Parallel()
779 f := newViewFixture(t, true, true)
780 v := f.view()
781 if v.Snapshot() != nil {
782 t.Fatal("a view that never published has a snapshot")
783 }
784 e1 := mustRebuild(t, v)
785 f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
786 e2 := mustRefresh(t, v)
787 if e2.Epoch <= e1.Epoch {
788 t.Fatalf("refresh epoch %d after %d", e2.Epoch, e1.Epoch)
789 }
790 if again := mustRefresh(t, v); again != e2 {
791 t.Fatal("a refresh that read nothing published a new snapshot")
792 }
793 e3 := mustRebuild(t, v)
794 if e3.Epoch <= e2.Epoch {
795 t.Fatalf("rebuild epoch %d after %d", e3.Epoch, e2.Epoch)
796 }
797}
798
799// The backstop holds mu for its whole walk, so an incremental refresh that
800// starts during it waits, and runs after the rebuild publishes: the rebuild
801// cannot publish over what the refresh read. Serial: it sets a package hook.
802func TestViewBackstopCannotOverwriteNewer(t *testing.T) {
803 f := newViewFixture(t, true, true)
804 v := f.view()
805 first := mustRebuild(t, v)
806
807 paused, resume := make(chan struct{}), make(chan struct{})
808 var once sync.Once
809 recordViewRebuildHook = func() {
810 once.Do(func() {
811 close(paused)
812 <-resume
813 })
814 }
815 t.Cleanup(func() { recordViewRebuildHook = nil })
816
817 rebuilt := make(chan error, 1)
818 go func() { rebuilt <- v.Rebuild(ctx) }()
819 <-paused
820 f.record(t, f.live.DirStore, f.acct[0], "shop", "three")
821 refreshed := make(chan error, 1)
822 go func() { refreshed <- v.Refresh(ctx) }()
823 select {
824 case err := <-refreshed:
825 t.Fatalf("a refresh ran while the backstop walk held the view: %v", err)
826 case <-time.After(100 * time.Millisecond):
827 }
828 close(resume)
829 if err := <-rebuilt; err != nil {
830 t.Fatal(err)
831 }
832 if err := <-refreshed; err != nil {
833 t.Fatal(err)
834 }
835 sn := v.Snapshot()
836 if sn.Epoch != first.Epoch+2 || !sn.Complete {
837 t.Fatalf("final epoch %d complete %v, want %d and complete", sn.Epoch, sn.Complete, first.Epoch+2)
838 }
839 if hosts := viewHosts(t, sn.Store(f.live)); !slices.Contains(hosts, "three.shop.example.com") {
840 t.Fatalf("the record written during the walk is missing: %v", hosts)
841 }
842}
843
844func TestSnapshotStoreReadYourWrites(t *testing.T) {
845 t.Parallel()
846 f := newViewFixture(t, true, true)
847 v := f.view()
848 sn := mustRebuild(t, v)
849 a := f.acct[0]
850 st := sn.Store(f.live)
851 rec := viewRecord(a, "fresh", "one")
852 if err := CreateRecord(ctx, st, rec); err != nil {
853 t.Fatal(err)
854 }
855 if got, err := GetRecord(ctx, st, a, "fresh", rec.Host); err != nil || got.Host != rec.Host {
856 t.Fatalf("Get after Put = %+v, %v", got, err)
857 }
858 if keys, err := st.List(ctx, RecordsPrefix(a, "fresh")); err != nil || !slices.Contains(keys, RecordKey(a, "fresh", rec.Host)) {
859 t.Fatalf("List after Put = %v, %v", keys, err)
860 }
861 if dirs, err := st.ListDirs(ctx, "apps/"+a+"/"); err != nil || !slices.Contains(dirs, "apps/"+a+"/fresh/") {
862 t.Fatalf("ListDirs after Put = %v, %v", dirs, err)
863 }
864 if err := PutManifest(ctx, st, a, "shop", Manifest{Schema: 1, App: "shop", Domain: "moved.example.com"}); err != nil {
865 t.Fatal(err)
866 }
867 if m, err := GetManifest(ctx, st, a, "shop"); err != nil || m.Domain != "moved.example.com" {
868 t.Fatalf("manifest after a CAS write = %+v, %v", m, err)
869 }
870 gone := RecordKey(a, "shop", "one.shop.example.com")
871 if err := st.Delete(ctx, gone); err != nil {
872 t.Fatal(err)
873 }
874 if _, _, err := st.Get(ctx, gone); !errors.Is(err, os.ErrNotExist) {
875 t.Fatalf("Get after Delete = %v", err)
876 }
877 if keys, _ := st.List(ctx, RecordsPrefix(a, "shop")); slices.Contains(keys, gone) {
878 t.Fatalf("List after Delete still has %s", gone)
879 }
880 // The snapshot itself never moves: a fresh store over it still answers
881 // from the cache.
882 if keys, _ := sn.Store(f.live).List(ctx, RecordsPrefix(a, "shop")); !slices.Contains(keys, gone) {
883 t.Fatal("a write through one snapshot store changed the snapshot")
884 }
885}
886
887// Seeded without the index, so the failed scope is not in it: only the
888// view's own record of the failure brings it back on the next Refresh.
889func TestSnapshotStoreErrorsMatchLive(t *testing.T) {
890 t.Parallel()
891 f := newViewFixture(t, true, false)
892 a, b := f.acct[0], f.acct[1]
893 f.live.setFailGet(ManifestKey(b, "docs"))
894 v := f.view()
895 sn := mustRebuild(t, v)
896 if sn.Complete {
897 t.Fatal("a snapshot holding a failed manifest read is complete")
898 }
899 st := sn.Store(f.live)
900 for _, key := range []string{
901 RecordKey(a, "shop", "missing.shop.example.com"),
902 RecordKey(a, "nowhere", "missing.nowhere.example.com"),
903 SidecarRecordKey(a, "shop", "missing"),
904 ManifestKey(b, "wiki"),
905 SidecarManifestKey(a, "blog"),
906 release.AppTombstoneKey(a, "shop"),
907 "accounts/zzzz/account.json",
908 } {
909 _, _, lerr := f.live.DirStore.Get(ctx, key)
910 _, _, serr := st.Get(ctx, key)
911 if !errors.Is(lerr, os.ErrNotExist) || !errors.Is(serr, os.ErrNotExist) {
912 t.Fatalf("Get %s: live %v, snapshot %v; want not-found from both", key, lerr, serr)
913 }
914 }
915 if _, err := GetManifest(ctx, st, b, "docs"); err == nil || errors.Is(err, os.ErrNotExist) {
916 t.Fatalf("a manifest whose read failed = %v, want the read error, not absence", err)
917 }
918 if tomb, err := release.ReadAppTombstone(ctx, st, a, "shop"); tomb != nil || err != nil {
919 t.Fatalf("absent trash marker = %+v, %v", tomb, err)
920 }
921 if tomb, err := release.ReadAppTombstone(ctx, st, a, "blog"); tomb == nil || err != nil {
922 t.Fatalf("trash marker = %+v, %v", tomb, err)
923 }
924
925 f.live.setFailGet("")
926 sn = mustRefresh(t, v)
927 if !sn.Complete {
928 t.Fatal("still incomplete after the manifest read recovered")
929 }
930 if m, err := GetManifest(ctx, sn.Store(f.live), b, "docs"); err != nil || m.App != "docs" {
931 t.Fatalf("recovered manifest = %+v, %v", m, err)
932 }
933}
934
935func TestSnapshotStoreOtherKeysGoLive(t *testing.T) {
936 t.Parallel()
937 f := newViewFixture(t, true, true)
938 v := f.view()
939 sn := mustRebuild(t, v)
940 a := f.acct[0]
941 if _, err := f.bypass.Put(ctx, "apps/"+a+"/shop/channels/edge.json", []byte(`{}`), ""); err != nil {
942 t.Fatal(err)
943 }
944 putReplace(t, f.bypass, "apps/"+a+"/shop/config.json", []byte(`{"after":true}`))
945 f.live.reset()
946 st := sn.Store(f.live)
947 body, _, err := st.Get(ctx, "apps/"+a+"/shop/config.json")
948 if err != nil || string(body) != `{"after":true}` {
949 t.Fatalf("config.json over the snapshot = %s, %v", body, err)
950 }
951 keys, err := st.List(ctx, "apps/"+a+"/shop/channels/")
952 if err != nil || !slices.Contains(keys, "apps/"+a+"/shop/channels/edge.json") {
953 t.Fatalf("channels over the snapshot = %v, %v", keys, err)
954 }
955 want := []string{"Get apps/" + a + "/shop/config.json", "List apps/" + a + "/shop/channels/"}
956 if got := f.live.got(); !reflect.DeepEqual(got, want) {
957 t.Fatalf("live saw %v, want %v", got, want)
958 }
959}
960
961func TestViewScopeApp(t *testing.T) {
962 t.Parallel()
963 for scope, want := range map[string][2]string{
964 "apps/abc/shop": {"abc", "shop"},
965 "apps/abc": {},
966 "apps/abc/": {},
967 "apps//shop": {},
968 "apps/a/b/c": {},
969 "accounts/abc": {},
970 } {
971 acct, app, ok := ScopeApp(scope)
972 if ok != (want[0] != "") || acct != want[0] || app != want[1] {
973 t.Fatalf("ScopeApp(%q) = %q, %q, %v", scope, acct, app, ok)
974 }
975 }
976}
977
978// A pass over a snapshot keeps the hold write's re-read fresh: the record is
979// re-read from the bucket, not from the snapshot the listing came from, so a
980// move that went on after the view was refreshed is not held again (the
981// guard TestReconcileHoldRereadsTheRecordBeforeHoldingARow pins on the walk).
982func TestSnapshotHoldRereadReadsTheBucket(t *testing.T) {
983 s, sqid := setup(t)
984 rec := movingRecord(sqid)
985 if err := CreateRecord(ctx, s, rec); err != nil {
986 t.Fatal(err)
987 }
988 v := NewRecordView(s, release.NewChangeWatch(s))
989 if err := v.Rebuild(ctx); err != nil {
990 t.Fatal(err)
991 }
992 committed := rec
993 m := *rec.Move
994 m.State = MoveCommitted
995 committed.Move = &m
996 key := RecordKey(rec.Account, rec.App, rec.Host)
997 _, etag, err := s.Get(ctx, key)
998 if err != nil {
999 t.Fatal(err)
1000 }
1001 if _, err := s.Put(ctx, key, mustJSON(t, committed), etag); err != nil {
1002 t.Fatal(err)
1003 }
1004 reg := regWith(t, RouteFor(rec, "/data", `["seed"]`, false, nil)) // the target's row, opened
1005 var wrote holdLog
1006 cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data", HoldWritten: wrote.written,
1007 Hold: func(r Record) (string, string, bool) { return "landing", r.Move.ID, r.Move.State == MoveLanding }}
1008 if err := Reconcile(ctx, v.Snapshot().Store(s), reg, cfg); err != nil {
1009 t.Fatal(err)
1010 }
1011 if got := mustGet(t, reg, rec.Host); got.Status != "ready" || got.HoldOp != "" || len(wrote.hosts) != 0 {
1012 t.Fatalf("an opened row was held again from the snapshot's record: %+v %v", got, wrote.hosts)
1013 }
1014}
1015
1016// staleView builds a view over s and returns it Complete, so that writes
1017// made after it, with no bump (s has no change index: an old binary, or a
1018// bump that failed), are invisible to its snapshots until the backstop.
1019func staleView(t *testing.T, s release.Store) *RecordView {
1020 t.Helper()
1021 v := NewRecordView(s, release.NewChangeWatch(s))
1022 if err := v.Refresh(ctx); err != nil {
1023 t.Fatal(err)
1024 }
1025 return v
1026}
1027
1028func completeSnapshot(t *testing.T, v *RecordView) *ViewSnapshot {
1029 t.Helper()
1030 if err := v.Refresh(ctx); err != nil {
1031 t.Fatal(err)
1032 }
1033 sn := v.Snapshot()
1034 if sn == nil || !sn.Complete {
1035 t.Fatal("fixture: the snapshot is not Complete")
1036 }
1037 return sn
1038}
1039
1040// A record written without its bump is absent from a Complete snapshot. A
1041// pass over that snapshot confirms the absence in the bucket before it
1042// retires the row a walking pass minted, and keeps it; a record that is
1043// really gone is still retired.
1044func TestSnapshotAbsenceDoesNotRetireALiveRecord(t *testing.T) {
1045 s, sqid := setup(t)
1046 v := staleView(t, s)
1047 rec := acmeRecord(sqid)
1048 if err := CreateRecord(ctx, s, rec); err != nil {
1049 t.Fatal(err)
1050 }
1051 reg := regWith(t)
1052 cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}
1053 if err := Reconcile(ctx, s, reg, cfg); err != nil { // a walking pass (the view could not vouch)
1054 t.Fatal(err)
1055 }
1056 mustGet(t, reg, rec.Host)
1057 sn := completeSnapshot(t, v)
1058 if recs, err := ListAllRecords(ctx, sn.Store(s)); err != nil || len(recs) != 0 {
1059 t.Fatalf("fixture: the snapshot has records %v, %v; it should lack the unbumped one", recs, err)
1060 }
1061 if err := Reconcile(ctx, sn.Store(s), reg, cfg); err != nil {
1062 t.Fatal(err)
1063 }
1064 mustGet(t, reg, rec.Host) // kept: the bucket has the record
1065
1066 if err := DeleteRecord(ctx, s, sqid, "tito", rec.Host); err != nil {
1067 t.Fatal(err)
1068 }
1069 if err := Reconcile(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
1070 t.Fatal(err)
1071 }
1072 if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
1073 t.Fatalf("a record really gone did not retire its row over the snapshot: %+v %v", rt, err)
1074 }
1075}
1076
1077// The same for the trash: a marker the snapshot still has, taken away
1078// without a bump, does not retire the app's rows.
1079func TestSnapshotTrashMarkerIsConfirmedBeforeRemoval(t *testing.T) {
1080 _, pub := releaseKeys(t)
1081 s, sqid := setup(t)
1082 rec := acmeRecord(sqid)
1083 if err := CreateRecord(ctx, s, rec); err != nil {
1084 t.Fatal(err)
1085 }
1086 reg := regWith(t)
1087 cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data", ReleasePub: pub}
1088 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1089 t.Fatal(err)
1090 }
1091 mustGet(t, reg, rec.Host)
1092 if err := release.TrashApp(ctx, s, sqid, "tito", "owner@example.com", time.Now()); err != nil {
1093 t.Fatal(err)
1094 }
1095 v := staleView(t, s) // sees the marker
1096 if err := release.RestoreApp(ctx, s, sqid, "tito"); err != nil {
1097 t.Fatal(err)
1098 }
1099 if err := Reconcile(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
1100 t.Fatal(err)
1101 }
1102 mustGet(t, reg, rec.Host) // kept: the bucket has no marker any more
1103}
1104
1105func TestSnapshotAbsenceDoesNotRetireALiveSidecar(t *testing.T) {
1106 s := dirStore(t)
1107 sqid := writeAccountManifest(t, s, acctID, "Shop")
1108 if err := PutSidecarManifest(ctx, s, sqid, "shop", SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable"}); err != nil {
1109 t.Fatal(err)
1110 }
1111 v := staleView(t, s)
1112 host := SidecarHost("bot", "shop", sqid)
1113 if err := CreateSidecarRecord(ctx, s, SidecarRecord{Schema: 1, Name: "bot", Host: host, App: "shop", Account: sqid, Channel: "stable"}); err != nil {
1114 t.Fatal(err)
1115 }
1116 reg := regWith(t)
1117 cfg := ReconcileConfig{DataDir: t.TempDir()}
1118 if err := ReconcileSidecars(ctx, s, reg, cfg); err != nil {
1119 t.Fatal(err)
1120 }
1121 mustGet(t, reg, host)
1122 if err := ReconcileSidecars(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
1123 t.Fatal(err)
1124 }
1125 mustGet(t, reg, host) // kept: the bucket has the record
1126
1127 if err := s.Delete(ctx, SidecarRecordKey(sqid, "shop", "bot")); err != nil {
1128 t.Fatal(err)
1129 }
1130 if err := ReconcileSidecars(ctx, completeSnapshot(t, v).Store(s), reg, cfg); err != nil {
1131 t.Fatal(err)
1132 }
1133 if rt, err := reg.Get(host); err != nil || rt != nil {
1134 t.Fatalf("a sidecar record really gone did not retire its row over the snapshot: %+v %v", rt, err)
1135 }
1136}
1137
1138// removalWorld is one account with app shop (instances and sidecars
1139// manifests) over a bucket whose writes bump the change index (live), with a
1140// second handle on the same directory whose writes do not (bypass: a bump
1141// that failed, an old binary, a direct-bucket writer). v follows live.
1142type removalWorld struct {
1143 live *viewCountingStore
1144 bypass *release.DirStore
1145 acct string
1146 v *RecordView
1147}
1148
1149func newRemovalWorld(t *testing.T) *removalWorld {
1150 t.Helper()
1151 root := t.TempDir()
1152 d := release.NewDirStore(root)
1153 d.EnableChangeIndex()
1154 w := &removalWorld{live: &viewCountingStore{DirStore: d}, bypass: release.NewDirStore(root)}
1155 w.acct = writeAccountManifest(t, w.live, 401, "Shop")
1156 if err := PutManifest(ctx, w.live, w.acct, "shop", Manifest{Schema: 1, App: "shop", Domain: "example.com"}); err != nil {
1157 t.Fatal(err)
1158 }
1159 if err := PutSidecarManifest(ctx, w.live, w.acct, "shop", SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable"}); err != nil {
1160 t.Fatal(err)
1161 }
1162 w.v = NewRecordView(w.live, release.NewChangeWatch(w.live))
1163 return w
1164}
1165
1166// snap is the view's current Complete snapshot as a store over live.
1167func (w *removalWorld) snap(t *testing.T) release.Store {
1168 t.Helper()
1169 return completeSnapshot(t, w.v).Store(w.live)
1170}
1171
1172func (w *removalWorld) record(app, slug, region, channel string) Record {
1173 return Record{Schema: 1, Host: slug + ".example.com", Slug: slug, App: app, Account: w.acct,
1174 Region: region, Channel: channel, CreatedAt: time.Now()}
1175}
1176
1177// listings counts the record-family listings live answered since the last
1178// reset: what a walk costs and a snapshot pass must not.
1179func (w *removalWorld) listings() int {
1180 n := 0
1181 for _, c := range w.live.got() {
1182 for _, m := range []string{"List ", "ListDirs ", "ListETags "} {
1183 if strings.HasPrefix(c, m+"accounts/") || strings.HasPrefix(c, m+"apps/") {
1184 n++
1185 }
1186 }
1187 }
1188 return n
1189}
1190
1191// Final review findings 2 and 4 (amadan #336): a removal a snapshot proves
1192// by absence is decided by the walk's own rules over the live bucket. A
1193// sidecar whose placement becomes no-inbound, in its record or in the app's
1194// sidecars manifest, with the bump landing, leaves an inbound box: the live
1195// walk does not place it either. Before, the confirmation asked only region
1196// and pool, and kept the row for ever.
1197func TestSnapshotSidecarPlacementChangeEvacuates(t *testing.T) {
1198 for _, tc := range []struct {
1199 name string
1200 apply func(t *testing.T, w *removalWorld, rec SidecarRecord)
1201 }{
1202 {"record", func(t *testing.T, w *removalWorld, rec SidecarRecord) {
1203 rec.Placement = PlacementNoInbound
1204 putReplace(t, w.live, SidecarRecordKey(w.acct, "shop", "bot"), mustJSON(t, rec))
1205 }},
1206 {"manifest", func(t *testing.T, w *removalWorld, _ SidecarRecord) {
1207 putReplace(t, w.live, SidecarManifestKey(w.acct, "shop"), mustJSON(t,
1208 SidecarManifest{Schema: 1, App: "shop", Args: []string{"run"}, Channel: "stable", Placement: PlacementNoInbound}))
1209 }},
1210 } {
1211 t.Run(tc.name, func(t *testing.T) {
1212 t.Parallel()
1213 w := newRemovalWorld(t)
1214 host := SidecarHost("bot", "shop", w.acct)
1215 rec := SidecarRecord{Schema: 1, Name: "bot", Host: host, App: "shop", Account: w.acct, Channel: "stable"}
1216 if err := CreateSidecarRecord(ctx, w.live, rec); err != nil {
1217 t.Fatal(err)
1218 }
1219 reg := regWith(t)
1220 cfg := ReconcileConfig{DataDir: t.TempDir()} // an inbound box
1221 if err := ReconcileSidecars(ctx, w.snap(t), reg, cfg); err != nil {
1222 t.Fatal(err)
1223 }
1224 mustGet(t, reg, host)
1225 sn := w.snap(t)
1226 w.live.reset()
1227 if err := ReconcileSidecars(ctx, sn, reg, cfg); err != nil {
1228 t.Fatal(err)
1229 }
1230 if n := w.listings(); n != 0 {
1231 t.Fatalf("a snapshot pass with nothing to remove listed the bucket: %v", w.live.got())
1232 }
1233
1234 tc.apply(t, w, rec)
1235 if err := ReconcileSidecars(ctx, w.snap(t), reg, cfg); err != nil {
1236 t.Fatal(err)
1237 }
1238 if rt, err := reg.Get(host); err != nil || rt != nil {
1239 t.Fatalf("a no-inbound sidecar stayed on an inbound box over the snapshot: %+v %v", rt, err)
1240 }
1241 })
1242 }
1243}
1244
1245// A record the snapshot lacks, whose live copy has an empty channel and a
1246// region out of this pass's scope: the walk protects an unusable record
1247// before it asks about scope, so the row stays. Before, the confirmation
1248// decoded the record, found it out of scope, and removed the row.
1249func TestSnapshotRemovalProtectsAnInvalidOutOfScopeRecord(t *testing.T) {
1250 t.Parallel()
1251 w := newRemovalWorld(t)
1252 sn := w.snap(t) // before the record: it never reaches the view
1253 rec := w.record("shop", "one", "ie", "stable")
1254 if err := CreateRecord(ctx, w.bypass, rec); err != nil {
1255 t.Fatal(err)
1256 }
1257 reg := regWith(t)
1258 cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
1259 if err := Reconcile(ctx, w.live, reg, cfg); err != nil { // a walking pass mints
1260 t.Fatal(err)
1261 }
1262 mustGet(t, reg, rec.Host)
1263 bad := rec
1264 bad.Region, bad.Channel = "uk", ""
1265 putReplace(t, w.bypass, RecordKey(w.acct, "shop", rec.Host), mustJSON(t, bad))
1266 if err := Reconcile(ctx, sn, reg, cfg); err != nil {
1267 t.Fatal(err)
1268 }
1269 mustGet(t, reg, rec.Host) // protected, as the walk protects it
1270 if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
1271 t.Fatal(err)
1272 }
1273 mustGet(t, reg, rec.Host) // and the walk agrees
1274}
1275
1276// A host whose record moved to another key (another app here) without a
1277// bump: the walk matches hosts across the whole deployment, so the row
1278// stays. Before, the confirmation read only the row's old key, found it
1279// gone, and removed the row. Once the record is gone everywhere, the row
1280// goes.
1281func TestSnapshotRemovalKeepsARehomedHost(t *testing.T) {
1282 t.Parallel()
1283 w := newRemovalWorld(t)
1284 w.snap(t) // built before the record: no bypass write below reaches the view
1285 rec := w.record("shop", "one", "ie", "stable")
1286 if err := CreateRecord(ctx, w.bypass, rec); err != nil {
1287 t.Fatal(err)
1288 }
1289 reg := regWith(t)
1290 cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
1291 if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
1292 t.Fatal(err)
1293 }
1294 mustGet(t, reg, rec.Host)
1295 if err := DeleteRecord(ctx, w.bypass, w.acct, "shop", rec.Host); err != nil {
1296 t.Fatal(err)
1297 }
1298 moved := w.record("cafe", "one", "ie", "stable")
1299 if err := CreateRecord(ctx, w.bypass, moved); err != nil {
1300 t.Fatal(err)
1301 }
1302 if err := Reconcile(ctx, w.snap(t), reg, cfg); err != nil {
1303 t.Fatal(err)
1304 }
1305 mustGet(t, reg, rec.Host) // kept: the bucket still has the host
1306
1307 if err := DeleteRecord(ctx, w.bypass, w.acct, "cafe", rec.Host); err != nil {
1308 t.Fatal(err)
1309 }
1310 if err := Reconcile(ctx, w.snap(t), reg, cfg); err != nil {
1311 t.Fatal(err)
1312 }
1313 if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
1314 t.Fatalf("a host really gone did not retire its row over the snapshot: %+v %v", rt, err)
1315 }
1316}
1317
1318// Disarmed (any store that is not a snapshot), removal is the walk's, from
1319// the one walk the pass already made: no second walk.
1320func TestRemovalOverTheBucketWalksOnce(t *testing.T) {
1321 t.Parallel()
1322 w := newRemovalWorld(t)
1323 rec := w.record("shop", "one", "ie", "stable")
1324 if err := CreateRecord(ctx, w.live, rec); err != nil {
1325 t.Fatal(err)
1326 }
1327 reg := regWith(t)
1328 cfg := ReconcileConfig{Region: "ie", DataDir: "/data"}
1329 if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
1330 t.Fatal(err)
1331 }
1332 mustGet(t, reg, rec.Host)
1333 if err := DeleteRecord(ctx, w.live, w.acct, "shop", rec.Host); err != nil {
1334 t.Fatal(err)
1335 }
1336 w.live.reset()
1337 if err := Reconcile(ctx, w.live, reg, cfg); err != nil {
1338 t.Fatal(err)
1339 }
1340 if rt, err := reg.Get(rec.Host); err != nil || rt != nil {
1341 t.Fatalf("a gone record's row survived a walking pass: %+v %v", rt, err)
1342 }
1343 n := 0
1344 for _, c := range w.live.got() {
1345 if c == "List accounts/" {
1346 n++
1347 }
1348 }
1349 if n != 1 {
1350 t.Fatalf("a walking pass listed accounts/ %d time(s), want 1: %v", n, w.live.got())
1351 }
1352}
1353