| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "os" |
| 7 | "path/filepath" |
| 8 | "strings" |
| 9 | "sync/atomic" |
| 10 | "testing" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/provision" |
| 14 | "github.com/carlosframework/platform/internal/release" |
| 15 | "github.com/carlosframework/platform/internal/seeds" |
| 16 | ) |
| 17 | |
| 18 | // sweepFixture is a console with a data bucket for seeds: a DirStore, so |
| 19 | // a snapshot's last-modified time is its file's mtime and a test can age |
| 20 | // it. Records are app shop's in the operator account. |
| 21 | type sweepFixture struct { |
| 22 | fx *consoleFixture |
| 23 | data release.Store |
| 24 | root string |
| 25 | } |
| 26 | |
| 27 | const sweepApp = "shop" |
| 28 | |
| 29 | func newSweepFixture(t *testing.T) sweepFixture { |
| 30 | t.Helper() |
| 31 | fx := newConsoleFixture(t) |
| 32 | root := t.TempDir() |
| 33 | data := release.NewDirStore(root) |
| 34 | fx.srv.cfg.SeedData = data |
| 35 | return sweepFixture{fx: fx, data: data, root: root} |
| 36 | } |
| 37 | |
| 38 | // seeded writes a record of shop on host whose seed is a fresh id, of |
| 39 | // kind, in state, stamped at, and answers the id. |
| 40 | func (f sweepFixture) seeded(t *testing.T, host, kind, state string, at time.Time) string { |
| 41 | t.Helper() |
| 42 | id := newSeedID(t) |
| 43 | f.recordOf(t, sweepApp, host, &provision.SeedSource{ID: id, Kind: kind, State: state, By: "bob@example.com", At: at}) |
| 44 | return id |
| 45 | } |
| 46 | |
| 47 | // recordOf writes a record of app on host with seed (nil for none). |
| 48 | func (f sweepFixture) recordOf(t *testing.T, app, host string, seed *provision.SeedSource) { |
| 49 | t.Helper() |
| 50 | rec := provision.Record{Schema: 1, Host: host, Slug: strings.Split(host, ".")[0], App: app, Account: release.OpsAccount, |
| 51 | Channel: "stable", CreatedAt: time.Now().UTC(), Seed: seed} |
| 52 | if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil { |
| 53 | t.Fatal(err) |
| 54 | } |
| 55 | } |
| 56 | |
| 57 | // snapshot writes seeds/<host>/<id>/snapshot.db last modified age ago. |
| 58 | func (f sweepFixture) snapshot(t *testing.T, host, id string, age time.Duration) string { |
| 59 | t.Helper() |
| 60 | key := seeds.SnapshotKey(host, id) |
| 61 | if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil { |
| 62 | t.Fatal(err) |
| 63 | } |
| 64 | // DirStore keeps a key at root/key, and its ListMod reports the file's |
| 65 | // mtime. |
| 66 | when := time.Now().Add(-age) |
| 67 | if err := os.Chtimes(filepath.Join(f.root, filepath.FromSlash(key)), when, when); err != nil { |
| 68 | t.Fatal(err) |
| 69 | } |
| 70 | return key |
| 71 | } |
| 72 | |
| 73 | func (f sweepFixture) snapshotExists(t *testing.T, key string) bool { |
| 74 | t.Helper() |
| 75 | _, _, err := f.data.Get(context.Background(), key) |
| 76 | if err != nil && !errors.Is(err, os.ErrNotExist) { |
| 77 | t.Fatal(err) |
| 78 | } |
| 79 | return err == nil |
| 80 | } |
| 81 | |
| 82 | // state is the seed state of shop's record on host, or "gone". |
| 83 | func (f sweepFixture) state(t *testing.T, host string) string { |
| 84 | t.Helper() |
| 85 | rec, err := provision.GetRecord(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host) |
| 86 | switch { |
| 87 | case errors.Is(err, os.ErrNotExist): |
| 88 | return "gone" |
| 89 | case err != nil: |
| 90 | t.Fatal(err) |
| 91 | } |
| 92 | return rec.Seed.State |
| 93 | } |
| 94 | |
| 95 | func (f sweepFixture) pass(now time.Time) { f.fx.srv.seedsPass(context.Background(), now) } |
| 96 | |
| 97 | func TestSweepLeavesALiveReservation(t *testing.T) { |
| 98 | f := newSweepFixture(t) |
| 99 | now := time.Now() |
| 100 | id := f.seeded(t, "canary.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-10*time.Minute)) |
| 101 | key := f.snapshot(t, "canary.shop.example.com", id, 10*time.Minute) |
| 102 | f.pass(now) |
| 103 | if got := f.state(t, "canary.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, key) { |
| 104 | t.Fatalf("a live reservation was touched: %s, snapshot kept %v", got, f.snapshotExists(t, key)) |
| 105 | } |
| 106 | // Past the hour it is abandoned by CAS, and its snapshot goes with it. |
| 107 | f.pass(now.Add(2 * time.Hour)) |
| 108 | if got := f.state(t, "canary.shop.example.com"); got != seeds.StateAborted { |
| 109 | t.Fatalf("a stale reservation = %s, want aborted", got) |
| 110 | } |
| 111 | if f.snapshotExists(t, key) { |
| 112 | t.Fatal("the snapshot of a seed this pass abandoned survived") |
| 113 | } |
| 114 | } |
| 115 | |
| 116 | // A --from copy is bounded at about 52 minutes (30m restore, 60s, 20m |
| 117 | // upload): a reservation younger than the hour is never abandoned. |
| 118 | func TestSweepReservationThresholdOutlastsTheCopy(t *testing.T) { |
| 119 | if seedReserveTTL <= 30*time.Minute+time.Minute+20*time.Minute { |
| 120 | t.Fatalf("seedReserveTTL %s does not outlast the --from copy's bound", seedReserveTTL) |
| 121 | } |
| 122 | f := newSweepFixture(t) |
| 123 | now := time.Now() |
| 124 | f.seeded(t, "copy.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-55*time.Minute)) |
| 125 | f.pass(now) |
| 126 | if got := f.state(t, "copy.shop.example.com"); got != seeds.StateReserving { |
| 127 | t.Fatalf("a 55-minute reservation = %s, want reserving", got) |
| 128 | } |
| 129 | } |
| 130 | |
| 131 | func TestSweepAbandonsOnlyStaleImports(t *testing.T) { |
| 132 | f := newSweepFixture(t) |
| 133 | now := time.Now() |
| 134 | old, young := now.Add(-8*24*time.Hour), now.Add(-6*24*time.Hour) |
| 135 | f.seeded(t, "waiting.shop.example.com", seeds.KindImport, seeds.StateAwaitingUpload, old) |
| 136 | f.seeded(t, "uploaded.shop.example.com", seeds.KindImport, seeds.StateUploaded, old) |
| 137 | f.seeded(t, "recent.shop.example.com", seeds.KindImport, seeds.StateUploaded, young) |
| 138 | f.seeded(t, "ready.shop.example.com", seeds.KindImport, seeds.StateReady, old) |
| 139 | f.seeded(t, "refused.shop.example.com", seeds.KindImport, seeds.StateRefused, old) |
| 140 | f.seeded(t, "fresh.shop.example.com", seeds.KindReplica, seeds.StateReady, old) |
| 141 | f.pass(now) |
| 142 | for host, want := range map[string]string{ |
| 143 | "waiting.shop.example.com": seeds.StateAborted, |
| 144 | "uploaded.shop.example.com": seeds.StateAborted, |
| 145 | "recent.shop.example.com": seeds.StateUploaded, |
| 146 | "ready.shop.example.com": seeds.StateReady, |
| 147 | "refused.shop.example.com": seeds.StateRefused, |
| 148 | "fresh.shop.example.com": seeds.StateReady, |
| 149 | } { |
| 150 | if got := f.state(t, host); got != want { |
| 151 | t.Errorf("%s = %s, want %s", host, got, want) |
| 152 | } |
| 153 | } |
| 154 | } |
| 155 | |
| 156 | func TestSweepDeletesOnlyUnreferencedOrSupersededSnapshots(t *testing.T) { |
| 157 | f := newSweepFixture(t) |
| 158 | now := time.Now() |
| 159 | keep, gone := map[string]string{}, map[string]string{} |
| 160 | |
| 161 | live := f.seeded(t, "a.shop.example.com", seeds.KindReplica, seeds.StateReady, now) |
| 162 | keep["a snapshot a live record needs"] = f.snapshot(t, "a.shop.example.com", live, 3*time.Hour) |
| 163 | gone["an orphan over an hour old beside a live seed"] = f.snapshot(t, "a.shop.example.com", newSeedID(t), 3*time.Hour) |
| 164 | keep["a young unreferenced snapshot (its door may be between record and upload)"] = |
| 165 | f.snapshot(t, "nobody.shop.example.com", newSeedID(t), 10*time.Minute) |
| 166 | gone["an orphan with no record, over an hour old"] = f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour) |
| 167 | |
| 168 | f.seeded(t, "b.shop.example.com", seeds.KindReplica, seeds.StatePlaced, now) |
| 169 | gone["a young snapshot superseded by a placed seed"] = f.snapshot(t, "b.shop.example.com", newSeedID(t), 5*time.Minute) |
| 170 | f.seeded(t, "c.shop.example.com", seeds.KindImport, seeds.StateCommitted, now) |
| 171 | gone["a young snapshot superseded by a committed seed"] = f.snapshot(t, "c.shop.example.com", newSeedID(t), 5*time.Minute) |
| 172 | f.seeded(t, "d.shop.example.com", seeds.KindImport, seeds.StateAborted, now) |
| 173 | keep["a young snapshot beside another seed that is not placed"] = f.snapshot(t, "d.shop.example.com", newSeedID(t), 5*time.Minute) |
| 174 | |
| 175 | own := func(host, kind, state string) string { |
| 176 | return f.snapshot(t, host, f.seeded(t, host, kind, state, now), 5*time.Minute) |
| 177 | } |
| 178 | gone["its own placed seed's (the box's delete was lost)"] = own("e.shop.example.com", seeds.KindReplica, seeds.StatePlaced) |
| 179 | gone["its own committed seed's"] = own("f.shop.example.com", seeds.KindImport, seeds.StateCommitted) |
| 180 | gone["its own aborted seed's"] = own("g.shop.example.com", seeds.KindImport, seeds.StateAborted) |
| 181 | keep["its own refused seed's"] = own("h.shop.example.com", seeds.KindImport, seeds.StateRefused) |
| 182 | keep["its own swapped seed's"] = own("i.shop.example.com", seeds.KindImport, seeds.StateSwapped) |
| 183 | keep["its own ready import's"] = own("j.shop.example.com", seeds.KindImport, seeds.StateReady) |
| 184 | |
| 185 | // Not a snapshot: nothing else under seeds/ is the sweep's. |
| 186 | stray := seeds.SnapshotPrefix("k.shop.example.com", newSeedID(t)) + "other.db" |
| 187 | if _, err := f.data.Put(context.Background(), stray, []byte("x"), ""); err != nil { |
| 188 | t.Fatal(err) |
| 189 | } |
| 190 | keep["an object that is not a snapshot"] = stray |
| 191 | |
| 192 | f.pass(now) |
| 193 | for why, key := range keep { |
| 194 | if !f.snapshotExists(t, key) { |
| 195 | t.Errorf("deleted %s", why) |
| 196 | } |
| 197 | } |
| 198 | for why, key := range gone { |
| 199 | if f.snapshotExists(t, key) { |
| 200 | t.Errorf("kept %s", why) |
| 201 | } |
| 202 | } |
| 203 | if got := f.state(t, "g.shop.example.com"); got != seeds.StateAborted { |
| 204 | t.Errorf("the aborted record = %s; the sweep deletes its snapshot and leaves the record to the box", got) |
| 205 | } |
| 206 | } |
| 207 | |
| 208 | func TestSweepKeepsSnapshotsOfAnUnreadableRecord(t *testing.T) { |
| 209 | f := newSweepFixture(t) |
| 210 | host := "a.shop.example.com" |
| 211 | id := f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, time.Now()) |
| 212 | key := f.snapshot(t, host, id, 3*time.Hour) |
| 213 | orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour) |
| 214 | rk := provision.RecordKey(release.OpsAccount, sweepApp, host) |
| 215 | _, etag, err := f.fx.rel.Get(context.Background(), rk) |
| 216 | if err != nil { |
| 217 | t.Fatal(err) |
| 218 | } |
| 219 | if _, err := f.fx.rel.Put(context.Background(), rk, []byte("{not json"), etag); err != nil { |
| 220 | t.Fatal(err) |
| 221 | } |
| 222 | f.pass(time.Now()) |
| 223 | if !f.snapshotExists(t, key) || !f.snapshotExists(t, orphan) { |
| 224 | t.Fatalf("an unreadable record's host lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, key), f.snapshotExists(t, orphan)) |
| 225 | } |
| 226 | } |
| 227 | |
| 228 | // Two records holding one host: neither decides for the other, so the |
| 229 | // sweep keeps everything of the host and abandons nothing on it. |
| 230 | func TestSweepKeepsAHostTwoRecordsHold(t *testing.T) { |
| 231 | f := newSweepFixture(t) |
| 232 | now := time.Now() |
| 233 | host := "twice.shop.example.com" |
| 234 | id := f.seeded(t, host, seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 235 | f.recordOf(t, "blog", host, &provision.SeedSource{ID: newSeedID(t), Kind: seeds.KindReplica, State: seeds.StatePlaced, |
| 236 | By: "bob@example.com", At: now}) |
| 237 | own := f.snapshot(t, host, id, 3*time.Hour) |
| 238 | orphan := f.snapshot(t, host, newSeedID(t), 3*time.Hour) |
| 239 | f.pass(now) |
| 240 | if !f.snapshotExists(t, own) || !f.snapshotExists(t, orphan) { |
| 241 | t.Fatalf("a host two records hold lost a snapshot: own kept %v, other kept %v", f.snapshotExists(t, own), f.snapshotExists(t, orphan)) |
| 242 | } |
| 243 | if got := f.state(t, host); got != seeds.StateReserving { |
| 244 | t.Fatalf("a host two records hold had its seed moved to %s", got) |
| 245 | } |
| 246 | } |
| 247 | |
| 248 | // The app stops the import between the abort's read and its write: the |
| 249 | // abort's CAS loses, the retry finds the seed moved on and refuses, and |
| 250 | // nothing is deleted. |
| 251 | func TestSweepAbandonsOnlyByCAS(t *testing.T) { |
| 252 | f := newSweepFixture(t) |
| 253 | host := "imp.shop.example.com" |
| 254 | old := time.Now().Add(-8 * 24 * time.Hour) |
| 255 | id := f.seeded(t, host, seeds.KindImport, seeds.StateUploaded, old) |
| 256 | key := f.snapshot(t, host, id, 8*24*time.Hour) |
| 257 | var calls []int |
| 258 | seedsPassBeforeAbort = func(call int) { |
| 259 | calls = append(calls, call) |
| 260 | if call != 1 { |
| 261 | return |
| 262 | } |
| 263 | if _, err := provision.SetRecordSeed(context.Background(), f.fx.rel, release.OpsAccount, sweepApp, host, provision.SeedStopped(id)); err != nil { |
| 264 | t.Errorf("the app's stop: %v", err) |
| 265 | } |
| 266 | } |
| 267 | t.Cleanup(func() { seedsPassBeforeAbort = nil }) |
| 268 | f.pass(time.Now()) |
| 269 | if len(calls) != 2 { |
| 270 | t.Fatalf("the abort's mutation ran %d times, want 2 (a write that lost, then a refused retry)", len(calls)) |
| 271 | } |
| 272 | if got := f.state(t, host); got != seeds.StateReady || !f.snapshotExists(t, key) { |
| 273 | t.Fatalf("state %s, snapshot kept %v; want the app's stop to win and nothing deleted", got, f.snapshotExists(t, key)) |
| 274 | } |
| 275 | } |
| 276 | |
| 277 | func TestSweepFinishesARolledBackSeed(t *testing.T) { |
| 278 | f := newSweepFixture(t) |
| 279 | host := "gone.shop.example.com" |
| 280 | id := f.seeded(t, host, seeds.KindImport, seeds.StateRolledBack, time.Now()) |
| 281 | key := f.snapshot(t, host, id, time.Minute) |
| 282 | f.pass(time.Now()) |
| 283 | if got := f.state(t, host); got != "gone" { |
| 284 | t.Fatalf("a rolled-back record after the sweep: %s", got) |
| 285 | } |
| 286 | if f.snapshotExists(t, key) { |
| 287 | t.Fatal("a finished seed's snapshot survived") |
| 288 | } |
| 289 | } |
| 290 | |
| 291 | // The host is deleted and seeded again between the two listings. Listed |
| 292 | // snapshots first, the new seed's snapshot is not in the listing, so the |
| 293 | // placed record of the old seed cannot condemn it (the doors write the |
| 294 | // record before the snapshot). |
| 295 | func TestSweepListsSnapshotsBeforeRecords(t *testing.T) { |
| 296 | f := newSweepFixture(t) |
| 297 | now := time.Now() |
| 298 | host := "again.shop.example.com" |
| 299 | f.seeded(t, host, seeds.KindReplica, seeds.StatePlaced, now.Add(-time.Hour)) |
| 300 | var fresh string |
| 301 | seedsPassBetweenListings = func() { |
| 302 | if err := f.fx.rel.Delete(context.Background(), provision.RecordKey(release.OpsAccount, sweepApp, host)); err != nil { |
| 303 | t.Error(err) |
| 304 | } |
| 305 | id := f.seeded(t, host, seeds.KindReplica, seeds.StateReady, now) |
| 306 | fresh = f.snapshot(t, host, id, time.Minute) |
| 307 | } |
| 308 | t.Cleanup(func() { seedsPassBetweenListings = nil }) |
| 309 | f.pass(now) |
| 310 | if fresh == "" { |
| 311 | t.Fatal("the seam never ran") |
| 312 | } |
| 313 | if !f.snapshotExists(t, fresh) { |
| 314 | t.Fatal("the new seed's snapshot was deleted as superseded") |
| 315 | } |
| 316 | } |
| 317 | |
| 318 | // failingAccountList fails the record walk's first listing. |
| 319 | type failingAccountList struct{ release.Store } |
| 320 | |
| 321 | func (s failingAccountList) List(ctx context.Context, prefix string) ([]string, error) { |
| 322 | if prefix == "accounts/" { |
| 323 | return nil, errors.New("the bucket is unreachable") |
| 324 | } |
| 325 | return s.Store.List(ctx, prefix) |
| 326 | } |
| 327 | |
| 328 | // failingModList is a data bucket whose listing fails. |
| 329 | type failingModList struct{ release.Store } |
| 330 | |
| 331 | func (failingModList) ListMod(context.Context, string) ([]release.KeyMod, error) { |
| 332 | return nil, errors.New("the bucket is unreachable") |
| 333 | } |
| 334 | |
| 335 | // A listing that fails is no knowledge: the pass deletes nothing and |
| 336 | // abandons nothing. |
| 337 | func TestSweepDecidesNothingOnAFailedListing(t *testing.T) { |
| 338 | for name, broke := range map[string]func(f sweepFixture){ |
| 339 | "records": func(f sweepFixture) { f.fx.srv.cfg.Releases = failingAccountList{f.fx.rel} }, |
| 340 | "snapshots": func(f sweepFixture) { f.fx.srv.cfg.SeedData = failingModList{f.data} }, |
| 341 | } { |
| 342 | t.Run(name, func(t *testing.T) { |
| 343 | f := newSweepFixture(t) |
| 344 | now := time.Now() |
| 345 | stale := f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 346 | staleKey := f.snapshot(t, "stale.shop.example.com", stale, 2*time.Hour) |
| 347 | orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour) |
| 348 | f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now) |
| 349 | broke(f) |
| 350 | f.pass(now) |
| 351 | if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving { |
| 352 | t.Errorf("a stale reservation = %s after a failed listing", got) |
| 353 | } |
| 354 | if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack { |
| 355 | t.Errorf("a rolled-back record = %s after a failed listing", got) |
| 356 | } |
| 357 | if !f.snapshotExists(t, staleKey) || !f.snapshotExists(t, orphan) { |
| 358 | t.Error("a snapshot was deleted after a failed listing") |
| 359 | } |
| 360 | }) |
| 361 | } |
| 362 | } |
| 363 | |
| 364 | // noModList is a data bucket that cannot list last-modified times. |
| 365 | type noModList struct{ release.Store } |
| 366 | |
| 367 | // Without last-modified times no snapshot is judged, but stale seeds are |
| 368 | // still abandoned and rolled-back ones finished. |
| 369 | func TestSweepWithoutListModStillAbandonsAndFinishes(t *testing.T) { |
| 370 | f := newSweepFixture(t) |
| 371 | now := time.Now() |
| 372 | f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 373 | f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now) |
| 374 | orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour) |
| 375 | f.fx.srv.cfg.SeedData = noModList{f.data} |
| 376 | f.pass(now) |
| 377 | if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted { |
| 378 | t.Errorf("a stale reservation = %s, want aborted", got) |
| 379 | } |
| 380 | if got := f.state(t, "done.shop.example.com"); got != "gone" { |
| 381 | t.Errorf("a rolled-back record = %s, want gone", got) |
| 382 | } |
| 383 | if !f.snapshotExists(t, orphan) { |
| 384 | t.Error("an orphan was deleted with no last-modified time to judge it by") |
| 385 | } |
| 386 | } |
| 387 | |
| 388 | // A console with host claims armed runs no sweep: its record writes go |
| 389 | // through the claim, and every seed door refuses it. |
| 390 | func TestSweepOffOnAHostClaimConsole(t *testing.T) { |
| 391 | f := newSweepFixture(t) |
| 392 | now := time.Now() |
| 393 | f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 394 | orphan := f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour) |
| 395 | f.fx.srv.cfg.HostClaim = true |
| 396 | ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 397 | defer cancel() |
| 398 | f.fx.srv.StartSeedsPass(ctx) // returns at once |
| 399 | if ctx.Err() != nil { |
| 400 | t.Fatal("StartSeedsPass ran on a host-claim console") |
| 401 | } |
| 402 | f.pass(now) |
| 403 | if got := f.state(t, "stale.shop.example.com"); got != seeds.StateReserving || !f.snapshotExists(t, orphan) { |
| 404 | t.Fatalf("a host-claim console swept: %s, orphan kept %v", got, f.snapshotExists(t, orphan)) |
| 405 | } |
| 406 | } |
| 407 | |
| 408 | // hang blocks its first caller until ctx is done, as a stalled connection |
| 409 | // does; every later call goes through. |
| 410 | type hang struct{ done atomic.Bool } |
| 411 | |
| 412 | func (h *hang) once(ctx context.Context) error { |
| 413 | if h.done.Swap(true) { |
| 414 | return nil |
| 415 | } |
| 416 | <-ctx.Done() |
| 417 | return ctx.Err() |
| 418 | } |
| 419 | |
| 420 | // hangingRecordWalk stalls the record walk's first listing. |
| 421 | type hangingRecordWalk struct { |
| 422 | release.Store |
| 423 | h *hang |
| 424 | } |
| 425 | |
| 426 | // DeleteIfMatch keeps the finish's conditional record delete reachable |
| 427 | // through the wrapper. |
| 428 | func (s hangingRecordWalk) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| 429 | return release.DeleteIfMatch(ctx, s.Store, key, etag) |
| 430 | } |
| 431 | |
| 432 | func (s hangingRecordWalk) List(ctx context.Context, prefix string) ([]string, error) { |
| 433 | if prefix == "accounts/" { |
| 434 | if err := s.h.once(ctx); err != nil { |
| 435 | return nil, err |
| 436 | } |
| 437 | } |
| 438 | return s.Store.List(ctx, prefix) |
| 439 | } |
| 440 | |
| 441 | // hangingData stalls the data bucket's first snapshot listing (list) or |
| 442 | // its first delete (!list). |
| 443 | type hangingData struct { |
| 444 | release.Store |
| 445 | h *hang |
| 446 | list bool |
| 447 | } |
| 448 | |
| 449 | func (s hangingData) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| 450 | if s.list { |
| 451 | if err := s.h.once(ctx); err != nil { |
| 452 | return nil, err |
| 453 | } |
| 454 | } |
| 455 | return s.Store.(release.ModLister).ListMod(ctx, prefix) |
| 456 | } |
| 457 | |
| 458 | func (s hangingData) Delete(ctx context.Context, key string) error { |
| 459 | if !s.list { |
| 460 | if err := s.h.once(ctx); err != nil { |
| 461 | return err |
| 462 | } |
| 463 | } |
| 464 | return s.Store.Delete(ctx, key) |
| 465 | } |
| 466 | |
| 467 | // A store call that never answers ends the pass at seedsPassTimeout, with |
| 468 | // nothing decided on what it could not finish; the next pass does the |
| 469 | // work. The delete case cuts the pass after its decisions, midway through |
| 470 | // its deletes, before its finishes. |
| 471 | func TestSweepPassIsBoundedByATimeout(t *testing.T) { |
| 472 | old := seedsPassTimeout |
| 473 | seedsPassTimeout = 100 * time.Millisecond |
| 474 | t.Cleanup(func() { seedsPassTimeout = old }) |
| 475 | for name, stall := range map[string]func(f sweepFixture){ |
| 476 | "record walk": func(f sweepFixture) { f.fx.srv.cfg.Releases = hangingRecordWalk{f.fx.rel, &hang{}} }, |
| 477 | "snapshot listing": func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, true} }, |
| 478 | "a snapshot delete": func(f sweepFixture) { f.fx.srv.cfg.SeedData = hangingData{f.data, &hang{}, false} }, |
| 479 | } { |
| 480 | t.Run(name, func(t *testing.T) { |
| 481 | f := newSweepFixture(t) |
| 482 | now := time.Now() |
| 483 | f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 484 | orphans := []string{ |
| 485 | f.snapshot(t, "left.shop.example.com", newSeedID(t), 3*time.Hour), |
| 486 | f.snapshot(t, "right.shop.example.com", newSeedID(t), 3*time.Hour), |
| 487 | } |
| 488 | f.seeded(t, "done.shop.example.com", seeds.KindImport, seeds.StateRolledBack, now) |
| 489 | stall(f) |
| 490 | done := make(chan struct{}) |
| 491 | go func() { f.pass(now); close(done) }() |
| 492 | select { |
| 493 | case <-done: |
| 494 | case <-time.After(10 * time.Second): |
| 495 | t.Fatal("a hung store call stalled the pass") |
| 496 | } |
| 497 | if got := f.state(t, "done.shop.example.com"); got != seeds.StateRolledBack { |
| 498 | t.Fatalf("a pass cut short finished a seed after its cut: %s", got) |
| 499 | } |
| 500 | f.pass(now) |
| 501 | if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted { |
| 502 | t.Errorf("the next pass left the stale reservation %s", got) |
| 503 | } |
| 504 | if got := f.state(t, "done.shop.example.com"); got != "gone" { |
| 505 | t.Errorf("the next pass left the rolled-back record %s", got) |
| 506 | } |
| 507 | for _, key := range orphans { |
| 508 | if f.snapshotExists(t, key) { |
| 509 | t.Errorf("the next pass left the orphan %s", key) |
| 510 | } |
| 511 | } |
| 512 | }) |
| 513 | } |
| 514 | } |
| 515 | |
| 516 | // M6's rolled-back half on its own: a finish that cannot complete (a store |
| 517 | // this seed made, and no backend to tear it down with) leaves the record |
| 518 | // rolled-back, and the sweep still deletes the seed's snapshot. |
| 519 | func TestSweepDeletesARolledBackSeedsSnapshotWhenItsFinishWaits(t *testing.T) { |
| 520 | f := finishFixture(t) |
| 521 | id := newSeedID(t) |
| 522 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 523 | f.markedStore(t, id) |
| 524 | f.fx.srv.cfg.StoreProvisioner = nil |
| 525 | key := seeds.SnapshotKey(fromHost, id) |
| 526 | if _, err := f.data.Put(context.Background(), key, []byte("db"), ""); err != nil { |
| 527 | t.Fatal(err) |
| 528 | } |
| 529 | f.fx.srv.seedsPass(context.Background(), time.Now()) |
| 530 | if _, _, err := f.data.Get(context.Background(), key); !errors.Is(err, os.ErrNotExist) { |
| 531 | t.Fatalf("the rolled-back seed's snapshot: %v, want deleted", err) |
| 532 | } |
| 533 | rec, err := f.record(t, fromHost) |
| 534 | if err != nil || rec.Seed.State != seeds.StateRolledBack { |
| 535 | t.Fatalf("the record after a finish that could not complete: %+v, %v; want it kept rolled-back", rec, err) |
| 536 | } |
| 537 | } |
| 538 | |
| 539 | // recordWalks keeps the listings a record walk makes (accounts/ and apps/). |
| 540 | func recordWalks(lists []string) []string { |
| 541 | var out []string |
| 542 | for _, l := range lists { |
| 543 | _, prefix, _ := strings.Cut(l, " ") |
| 544 | if strings.HasPrefix(prefix, "accounts/") || strings.HasPrefix(prefix, "apps/") { |
| 545 | out = append(out, l) |
| 546 | } |
| 547 | } |
| 548 | return out |
| 549 | } |
| 550 | |
| 551 | // Armed, a pass whose snapshot listing is empty has nothing to delete, so |
| 552 | // it makes no strict walk of the bucket: it reads the records from the |
| 553 | // record view, and a stale seed is still abandoned from what the view holds. |
| 554 | func TestSeedSweepSkipsWalkWithoutSnapshots(t *testing.T) { |
| 555 | t.Parallel() |
| 556 | f := newSweepFixture(t) |
| 557 | cs := indexedDeployment(t, f.fx) |
| 558 | now := time.Now() |
| 559 | f.seeded(t, "stale.shop.example.com", seeds.KindReplica, seeds.StateReserving, now.Add(-2*time.Hour)) |
| 560 | armRecordView(t, f.fx.srv, cs) |
| 561 | cs.taken() |
| 562 | f.pass(now) |
| 563 | if walks := recordWalks(cs.taken()); len(walks) != 0 { |
| 564 | t.Fatalf("a pass with no snapshot walked the records: %q", walks) |
| 565 | } |
| 566 | if got := f.state(t, "stale.shop.example.com"); got != seeds.StateAborted { |
| 567 | t.Fatalf("a stale reservation the view holds = %s, want aborted", got) |
| 568 | } |
| 569 | } |
| 570 | |
| 571 | // Armed, a pass that lists a snapshot keeps today's strict walk of the |
| 572 | // bucket, after the snapshot listing: a record written between the two |
| 573 | // listings that the view has not seen (here, a write that bumps no scope) |
| 574 | // still names its snapshot, so the snapshot is kept. |
| 575 | func TestSeedSweepWalksStrictWithSnapshots(t *testing.T) { |
| 576 | f := newSweepFixture(t) |
| 577 | bare := release.NewDirStore(f.fx.rel.(*deletableStore).root) // no change index: its writes bump nothing |
| 578 | cs := indexedDeployment(t, f.fx) |
| 579 | now := time.Now() |
| 580 | host := "late.shop.example.com" |
| 581 | id := newSeedID(t) |
| 582 | key := f.snapshot(t, host, id, 3*time.Hour) |
| 583 | armRecordView(t, f.fx.srv, cs) |
| 584 | seedsPassBetweenListings = func() { |
| 585 | cs.taken() |
| 586 | rec := provision.Record{Schema: 1, Host: host, Slug: "late", App: sweepApp, Account: release.OpsAccount, |
| 587 | Channel: "stable", CreatedAt: now.UTC(), |
| 588 | Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "bob@example.com", At: now}} |
| 589 | if err := provision.CreateRecord(context.Background(), bare, rec); err != nil { |
| 590 | t.Error(err) |
| 591 | } |
| 592 | } |
| 593 | t.Cleanup(func() { seedsPassBetweenListings = nil }) |
| 594 | f.pass(now) |
| 595 | if walks := recordWalks(cs.taken()); len(walks) == 0 { |
| 596 | t.Fatal("a pass with a snapshot made no record walk after the snapshot listing") |
| 597 | } |
| 598 | if !f.snapshotExists(t, key) { |
| 599 | t.Fatal("the snapshot a record names was deleted: the pass read the view, not the bucket") |
| 600 | } |
| 601 | } |
| 602 | |
amadan