| 1 | package provision |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "sync" |
| 9 | "testing" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/release" |
| 13 | ) |
| 14 | |
| 15 | // overlapStore holds each record Get open briefly and records the most Gets |
| 16 | // it saw in flight at once. |
| 17 | type overlapStore struct { |
| 18 | release.Store |
| 19 | mu sync.Mutex |
| 20 | inFlight int |
| 21 | maxFlight int |
| 22 | reads int |
| 23 | failKey string |
| 24 | } |
| 25 | |
| 26 | func (s *overlapStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 27 | if strings.Contains(key, "/instances/") { |
| 28 | s.mu.Lock() |
| 29 | s.inFlight++ |
| 30 | s.reads++ |
| 31 | if s.inFlight > s.maxFlight { |
| 32 | s.maxFlight = s.inFlight |
| 33 | } |
| 34 | s.mu.Unlock() |
| 35 | time.Sleep(20 * time.Millisecond) |
| 36 | s.mu.Lock() |
| 37 | s.inFlight-- |
| 38 | s.mu.Unlock() |
| 39 | if key == s.failKey { |
| 40 | return nil, "", errors.New("read refused") |
| 41 | } |
| 42 | } |
| 43 | return s.Store.Get(ctx, key) |
| 44 | } |
| 45 | |
| 46 | func seedRecords(t *testing.T, n int) (release.Store, string) { |
| 47 | t.Helper() |
| 48 | inner, sqid := setup(t) |
| 49 | for i := 0; i < n; i++ { |
| 50 | rec := acmeRecord(sqid) |
| 51 | slug := "r" + string(rune('a'+i)) |
| 52 | rec.Host, rec.Slug = slug+".go.tito.io", slug |
| 53 | if err := CreateRecord(ctx, inner, rec); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | } |
| 57 | return inner, sqid |
| 58 | } |
| 59 | |
| 60 | // The record readers read records several at a time. One at a time, a few |
| 61 | // hundred records cost the fast loop most of a 27-second pass on every |
| 62 | // serial bump (#231). |
| 63 | func TestRecordReadersReadRecordsConcurrently(t *testing.T) { |
| 64 | inner, _ := seedRecords(t, 12) |
| 65 | cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| 66 | for _, c := range []struct { |
| 67 | name string |
| 68 | run func(release.Store) error |
| 69 | }{ |
| 70 | {"Reconcile", func(s release.Store) error { return Reconcile(ctx, s, regWith(t), cfg) }}, |
| 71 | {"ListAllRecords", func(s release.Store) error { _, err := ListAllRecords(ctx, s); return err }}, |
| 72 | } { |
| 73 | s := &overlapStore{Store: inner} |
| 74 | if err := c.run(s); err != nil { |
| 75 | t.Fatalf("%s: %v", c.name, err) |
| 76 | } |
| 77 | if s.maxFlight < 2 || s.maxFlight > 8 { |
| 78 | t.Fatalf("%s read up to %d records at a time, want several and never more than 8", c.name, s.maxFlight) |
| 79 | } |
| 80 | if s.reads != 12 { |
| 81 | t.Fatalf("%s made %d record reads for 12 records, want one each", c.name, s.reads) |
| 82 | } |
| 83 | } |
| 84 | } |
| 85 | |
| 86 | // A read that fails is still that record's failure, judged where it always |
| 87 | // was: reconcile keeps the routes it could not read, and mints the rest. |
| 88 | func TestReconcileStillJudgesAFailedReadPerRecord(t *testing.T) { |
| 89 | inner, sqid := seedRecords(t, 4) |
| 90 | cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| 91 | reg := regWith(t) |
| 92 | if err := Reconcile(ctx, inner, reg, cfg); err != nil { |
| 93 | t.Fatal(err) |
| 94 | } |
| 95 | s := &overlapStore{Store: inner, failKey: RecordKey(sqid, "tito", "rb.go.tito.io")} |
| 96 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 97 | t.Fatal(err) |
| 98 | } |
| 99 | // A record created since the first pass must still be minted in a pass |
| 100 | // where another record's read fails: the failure is rb's alone. |
| 101 | fresh := acmeRecord(sqid) |
| 102 | fresh.Host, fresh.Slug = "re.go.tito.io", "re" |
| 103 | if err := CreateRecord(ctx, inner, fresh); err != nil { |
| 104 | t.Fatal(err) |
| 105 | } |
| 106 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 107 | t.Fatal(err) |
| 108 | } |
| 109 | for _, h := range []string{"ra.go.tito.io", "rb.go.tito.io", "rc.go.tito.io", "rd.go.tito.io", "re.go.tito.io"} { |
| 110 | if rt, err := reg.Get(h); err != nil || rt == nil { |
| 111 | t.Fatalf("%s missing after a pass where only rb's read failed: %+v, %v", h, rt, err) |
| 112 | } |
| 113 | } |
| 114 | _, err := ListAllRecords(ctx, s) |
| 115 | if err == nil || !strings.Contains(err.Error(), "rb.go.tito.io") { |
| 116 | t.Fatalf("ListAllRecords with rb unreadable = %v, want an error naming rb", err) |
| 117 | } |
| 118 | } |
| 119 | |
| 120 | // A loop that stops at its first bad read has read at most one chunk ahead, |
| 121 | // and a large record set is never prefetched whole. |
| 122 | func TestListAllRecordsStopsWithinAChunkOfABadRead(t *testing.T) { |
| 123 | inner, sqid := setup(t) |
| 124 | const total = 200 // more than one chunk, fixed so the test does not move with readChunk |
| 125 | for i := 0; i < total; i++ { |
| 126 | rec := acmeRecord(sqid) |
| 127 | rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i) |
| 128 | if err := CreateRecord(ctx, inner, rec); err != nil { |
| 129 | t.Fatal(err) |
| 130 | } |
| 131 | } |
| 132 | s := &countingGets{Store: inner, failKey: RecordKey(sqid, "tito", "r0000.go.tito.io")} |
| 133 | if _, err := ListAllRecords(ctx, s); err == nil { |
| 134 | t.Fatal("ListAllRecords read past an unreadable first record without an error") |
| 135 | } |
| 136 | if s.n > readChunk || s.n >= total { |
| 137 | t.Fatalf("read %d of %d records before stopping at the first, want at most one chunk (%d)", s.n, total, readChunk) |
| 138 | } |
| 139 | } |
| 140 | |
| 141 | // A cancelled context reads nothing, and every key reports why. |
| 142 | func TestGetEachOnACancelledContextReadsNothing(t *testing.T) { |
| 143 | inner, sqid := setup(t) |
| 144 | s := &countingGets{Store: inner} |
| 145 | cctx, cancel := context.WithCancel(ctx) |
| 146 | cancel() |
| 147 | key := RecordKey(sqid, "tito", "acme.go.tito.io") |
| 148 | got := getEach(cctx, s, []string{key}) |
| 149 | if s.n != 0 { |
| 150 | t.Fatalf("made %d reads on a cancelled context", s.n) |
| 151 | } |
| 152 | if !errors.Is(got[key].err, context.Canceled) { |
| 153 | t.Fatalf("key reported %v, want context.Canceled", got[key].err) |
| 154 | } |
| 155 | } |
| 156 | |
| 157 | type countingGets struct { |
| 158 | release.Store |
| 159 | mu sync.Mutex |
| 160 | n int |
| 161 | failKey string |
| 162 | } |
| 163 | |
| 164 | func (s *countingGets) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 165 | if strings.Contains(key, "/instances/") { |
| 166 | s.mu.Lock() |
| 167 | s.n++ |
| 168 | s.mu.Unlock() |
| 169 | if key == s.failKey { |
| 170 | return nil, "", errors.New("read refused") |
| 171 | } |
| 172 | } |
| 173 | return s.Store.Get(ctx, key) |
| 174 | } |
| 175 | |
| 176 | // appOverlapStore holds each app-level read (an app's instances manifest and |
| 177 | // its trash marker) open briefly and records the most in flight at once. |
| 178 | type appOverlapStore struct { |
| 179 | release.Store |
| 180 | mu sync.Mutex |
| 181 | inFlight int |
| 182 | maxFlight int |
| 183 | reads map[string]int |
| 184 | } |
| 185 | |
| 186 | func (s *appOverlapStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 187 | if strings.HasSuffix(key, "/instances.json") || strings.HasSuffix(key, "/deleted.json") { |
| 188 | s.mu.Lock() |
| 189 | s.inFlight++ |
| 190 | if s.inFlight > s.maxFlight { |
| 191 | s.maxFlight = s.inFlight |
| 192 | } |
| 193 | if s.reads == nil { |
| 194 | s.reads = map[string]int{} |
| 195 | } |
| 196 | s.reads[key]++ |
| 197 | s.mu.Unlock() |
| 198 | time.Sleep(20 * time.Millisecond) |
| 199 | s.mu.Lock() |
| 200 | s.inFlight-- |
| 201 | s.mu.Unlock() |
| 202 | } |
| 203 | return s.Store.Get(ctx, key) |
| 204 | } |
| 205 | |
| 206 | // Reconcile reads each app's manifest and trash marker several apps at a |
| 207 | // time, and each one once. |
| 208 | func TestReconcileReadsAppStateConcurrentlyAndOnce(t *testing.T) { |
| 209 | inner, sqid := setup(t) |
| 210 | want := map[string]bool{} |
| 211 | for i := 0; i < 10; i++ { |
| 212 | app := "app" + string(rune('a'+i)) |
| 213 | want[release.AppPrefix(sqid, app)+"instances.json"] = true |
| 214 | want[release.AppTombstoneKey(sqid, app)] = true |
| 215 | if err := PutManifest(ctx, inner, sqid, app, Manifest{Schema: 1, App: app, Domain: app + ".example", Pool: "tito"}); err != nil { |
| 216 | t.Fatal(err) |
| 217 | } |
| 218 | rec := acmeRecord(sqid) |
| 219 | rec.App, rec.Host, rec.Slug = app, "x."+app+".example", "x" |
| 220 | if err := CreateRecord(ctx, inner, rec); err != nil { |
| 221 | t.Fatal(err) |
| 222 | } |
| 223 | } |
| 224 | s := &appOverlapStore{Store: inner} |
| 225 | cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| 226 | if err := Reconcile(ctx, s, regWith(t), cfg); err != nil { |
| 227 | t.Fatal(err) |
| 228 | } |
| 229 | if s.maxFlight < 2 || s.maxFlight > 8 { |
| 230 | t.Fatalf("app state read up to %d at a time, want several and never more than 8", s.maxFlight) |
| 231 | } |
| 232 | for key := range want { |
| 233 | if s.reads[key] != 1 { |
| 234 | t.Fatalf("%s read %d times in one pass, want once", key, s.reads[key]) |
| 235 | } |
| 236 | } |
| 237 | } |
| 238 | |
| 239 | // An app in the trash is never minted for, so its manifest is not read: the |
| 240 | // order the record loop always asked in, trash first. |
| 241 | func TestReconcileDoesNotReadATrashedAppsManifest(t *testing.T) { |
| 242 | inner, sqid := setup(t) |
| 243 | tomb := fmt.Sprintf(`{"schema":1,"account":%q,"app":"tito","deleted_at":"2026-10-05T00:00:00Z","actor":"test"}`, sqid) |
| 244 | if _, err := inner.Put(ctx, release.AppTombstoneKey(sqid, "tito"), []byte(tomb), ""); err != nil { |
| 245 | t.Fatal(err) |
| 246 | } |
| 247 | if got, err := release.ReadAppTombstone(ctx, inner, sqid, "tito"); err != nil || got == nil { |
| 248 | t.Fatalf("fixture: the trash marker does not read as one: %+v, %v", got, err) |
| 249 | } |
| 250 | if err := CreateRecord(ctx, inner, acmeRecord(sqid)); err != nil { |
| 251 | t.Fatal(err) |
| 252 | } |
| 253 | s := &appOverlapStore{Store: inner} |
| 254 | cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| 255 | if err := Reconcile(ctx, s, regWith(t), cfg); err != nil { |
| 256 | t.Fatal(err) |
| 257 | } |
| 258 | if n := s.reads[release.AppPrefix(sqid, "tito")+"instances.json"]; n != 0 { |
| 259 | t.Fatalf("a trashed app's manifest was read %d times", n) |
| 260 | } |
| 261 | } |
| 262 | |
| 263 | // Every chunk is walked: 130 records, three chunks, each record minted and |
| 264 | // listed, each read once. |
| 265 | func TestRecordReadersWalkEveryChunk(t *testing.T) { |
| 266 | inner, sqid := setup(t) |
| 267 | const total = 130 |
| 268 | hosts := map[string]bool{} |
| 269 | for i := 0; i < total; i++ { |
| 270 | rec := acmeRecord(sqid) |
| 271 | rec.Host, rec.Slug = fmt.Sprintf("r%04d.go.tito.io", i), fmt.Sprintf("r%04d", i) |
| 272 | hosts[rec.Host] = true |
| 273 | if err := CreateRecord(ctx, inner, rec); err != nil { |
| 274 | t.Fatal(err) |
| 275 | } |
| 276 | } |
| 277 | s := &perKeyGets{Store: inner} |
| 278 | reg := regWith(t) |
| 279 | cfg := ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"} |
| 280 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 281 | t.Fatal(err) |
| 282 | } |
| 283 | for h := range hosts { |
| 284 | if rt, err := reg.Get(h); err != nil || rt == nil { |
| 285 | t.Fatalf("%s not minted: %+v, %v", h, rt, err) |
| 286 | } |
| 287 | if n := s.count(RecordKey(sqid, "tito", h)); n != 1 { |
| 288 | t.Fatalf("%s read %d times by Reconcile, want once", h, n) |
| 289 | } |
| 290 | } |
| 291 | recs, err := ListAllRecords(ctx, s) |
| 292 | if err != nil { |
| 293 | t.Fatal(err) |
| 294 | } |
| 295 | if len(recs) != total { |
| 296 | t.Fatalf("ListAllRecords = %d records, want %d", len(recs), total) |
| 297 | } |
| 298 | } |
| 299 | |
| 300 | type perKeyGets struct { |
| 301 | release.Store |
| 302 | mu sync.Mutex |
| 303 | n map[string]int |
| 304 | } |
| 305 | |
| 306 | func (s *perKeyGets) Get(ctx context.Context, key string) ([]byte, string, error) { |
| 307 | s.mu.Lock() |
| 308 | if s.n == nil { |
| 309 | s.n = map[string]int{} |
| 310 | } |
| 311 | s.n[key]++ |
| 312 | s.mu.Unlock() |
| 313 | return s.Store.Get(ctx, key) |
| 314 | } |
| 315 | |
| 316 | func (s *perKeyGets) count(key string) int { |
| 317 | s.mu.Lock() |
| 318 | defer s.mu.Unlock() |
| 319 | return s.n[key] |
| 320 | } |
| 321 | |
amadan