| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "strings" |
| 7 | "sync/atomic" |
| 8 | "testing" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/objstore" |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | var errTransientList = errors.New("transient listing failure") |
| 16 | |
| 17 | // flakyListStore fails listings under one prefix while failing is set. It |
| 18 | // embeds the concrete *release.DirStore so PageLister and ModLister stay |
| 19 | // promoted (see drainFixture). |
| 20 | type flakyListStore struct { |
| 21 | *release.DirStore |
| 22 | failing atomic.Bool |
| 23 | // dirsPrefix fails ListDirs(prefix) for exactly this prefix. |
| 24 | dirsPrefix string |
| 25 | // listPrefix fails List/ListPage for keys under this prefix. |
| 26 | listPrefix string |
| 27 | } |
| 28 | |
| 29 | func (f *flakyListStore) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| 30 | if f.failing.Load() && f.dirsPrefix != "" && prefix == f.dirsPrefix { |
| 31 | return nil, errTransientList |
| 32 | } |
| 33 | return f.DirStore.ListDirs(ctx, prefix) |
| 34 | } |
| 35 | |
| 36 | func (f *flakyListStore) List(ctx context.Context, prefix string) ([]string, error) { |
| 37 | if f.failing.Load() && f.listPrefix != "" && strings.HasPrefix(prefix, f.listPrefix) { |
| 38 | return nil, errTransientList |
| 39 | } |
| 40 | return f.DirStore.List(ctx, prefix) |
| 41 | } |
| 42 | |
| 43 | func (f *flakyListStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| 44 | if f.failing.Load() && f.listPrefix != "" && strings.HasPrefix(prefix, f.listPrefix) { |
| 45 | return nil, errTransientList |
| 46 | } |
| 47 | return f.DirStore.ListPage(ctx, prefix, startAfter, limit) |
| 48 | } |
| 49 | |
| 50 | func indexed(fx *consoleFixture, acct, app string) bool { |
| 51 | for _, ref := range fx.srv.scanQueue.Apps() { |
| 52 | if ref.Account == acct && ref.App == app { |
| 53 | return true |
| 54 | } |
| 55 | } |
| 56 | return false |
| 57 | } |
| 58 | |
| 59 | // TestRebuildRetriesAfterPartialFailure: a nested listing failure must make |
| 60 | // rebuildScanQueue return an error (so StartScanDrain retries) while the |
| 61 | // healthy app is still found, and the retry must find the skipped app. |
| 62 | func TestRebuildRetriesAfterPartialFailure(t *testing.T) { |
| 63 | const badAcct, goodAcct = "acct-bad", "acct-good" |
| 64 | cases := []struct { |
| 65 | name string |
| 66 | setup func(f *flakyListStore) |
| 67 | // acct/app of the skipped app, and a healthy app that must still be found |
| 68 | skipAcct, skipApp string |
| 69 | goodAcct, goodApp string |
| 70 | }{ |
| 71 | { |
| 72 | name: "account app-list failure", |
| 73 | setup: func(f *flakyListStore) { |
| 74 | f.dirsPrefix = "apps/" + badAcct + "/" |
| 75 | }, |
| 76 | skipAcct: badAcct, skipApp: "shop", |
| 77 | goodAcct: goodAcct, goodApp: "shop", |
| 78 | }, |
| 79 | { |
| 80 | name: "per-app queue-list failure", |
| 81 | setup: func(f *flakyListStore) { |
| 82 | f.listPrefix = objstore.QueuePrefix(badAcct, "shop") |
| 83 | }, |
| 84 | skipAcct: badAcct, skipApp: "shop", |
| 85 | goodAcct: badAcct, goodApp: "blog", |
| 86 | }, |
| 87 | } |
| 88 | for _, tc := range cases { |
| 89 | t.Run(tc.name, func(t *testing.T) { |
| 90 | ctx := context.Background() |
| 91 | fx, _, _ := drainFixture(t) |
| 92 | flaky := &flakyListStore{DirStore: fx.rel.(*release.DirStore)} |
| 93 | tc.setup(flaky) |
| 94 | fx.srv.cfg.Releases = flaky |
| 95 | |
| 96 | // Names sort so the failing account/app comes first. |
| 97 | enableStore(t, fx, tc.skipAcct, tc.skipApp, true) |
| 98 | enableStore(t, fx, tc.goodAcct, tc.goodApp, true) |
| 99 | at := time.Now().Add(-time.Minute) |
| 100 | enqueueStranded(t, fx, tc.skipAcct, tc.skipApp, "uploads/skipped.pdf", at) |
| 101 | enqueueStranded(t, fx, tc.goodAcct, tc.goodApp, "uploads/healthy.pdf", at) |
| 102 | |
| 103 | flaky.failing.Store(true) |
| 104 | err := fx.srv.rebuildScanQueue(ctx) |
| 105 | if !errors.Is(err, errTransientList) { |
| 106 | t.Fatalf("first rebuild err = %v, want the transient listing failure so the drain retries", err) |
| 107 | } |
| 108 | if !indexed(fx, tc.goodAcct, tc.goodApp) { |
| 109 | t.Errorf("healthy app %s/%s not indexed: one failure cost the others their queue", tc.goodAcct, tc.goodApp) |
| 110 | } |
| 111 | if indexed(fx, tc.skipAcct, tc.skipApp) { |
| 112 | t.Errorf("skipped app %s/%s indexed despite its listing failing", tc.skipAcct, tc.skipApp) |
| 113 | } |
| 114 | // Nothing was deleted or lost by the failed pass. |
| 115 | if n := queueLen(t, fx, tc.skipAcct, tc.skipApp); n != 1 { |
| 116 | t.Errorf("skipped app has %d queue item(s) after the failed pass, want 1", n) |
| 117 | } |
| 118 | if n := queueLen(t, fx, tc.goodAcct, tc.goodApp); n != 1 { |
| 119 | t.Errorf("healthy app has %d queue item(s) after the pass, want 1", n) |
| 120 | } |
| 121 | |
| 122 | flaky.failing.Store(false) |
| 123 | if err := fx.srv.rebuildScanQueue(ctx); err != nil { |
| 124 | t.Fatalf("second rebuild after the failure cleared: %v", err) |
| 125 | } |
| 126 | if !indexed(fx, tc.skipAcct, tc.skipApp) { |
| 127 | t.Errorf("retry did not find the previously skipped app %s/%s", tc.skipAcct, tc.skipApp) |
| 128 | } |
| 129 | if n := queueLen(t, fx, tc.skipAcct, tc.skipApp); n != 1 { |
| 130 | t.Errorf("skipped app has %d queue item(s) after the retry, want 1", n) |
| 131 | } |
| 132 | if d := fx.srv.ScanQueueDepth(); d != 2 { |
| 133 | t.Errorf("index depth %d after the retry, want 2", d) |
| 134 | } |
| 135 | }) |
| 136 | } |
| 137 | } |
| 138 | |
amadan