| 1 | package console |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "net/http/httptest" |
| 8 | "os" |
| 9 | "path/filepath" |
| 10 | "slices" |
| 11 | "strings" |
| 12 | "testing" |
| 13 | "time" |
| 14 | |
| 15 | "github.com/carlosframework/platform/internal/dbdoor/dbdoortest" |
| 16 | "github.com/carlosframework/platform/internal/lease" |
| 17 | "github.com/carlosframework/platform/internal/lease/leasetest" |
| 18 | "github.com/carlosframework/platform/internal/objstore" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/release" |
| 21 | "github.com/carlosframework/platform/internal/seeds" |
| 22 | ) |
| 23 | |
| 24 | // finishFixture is the --from fixture with private storage on and one data |
| 25 | // bucket, as production has it: seeds/ and every instance's replica in the |
| 26 | // same store, so the leftover-replica rail and the finish see one bucket. |
| 27 | // The host's lease is read by the lease package itself (lease.LockState), |
| 28 | // so it is served from leasetest's S3 server; leases answers for it. |
| 29 | func finishFixture(t *testing.T) fromFixture { |
| 30 | t.Helper() |
| 31 | f := newFromFixture(t, 3) |
| 32 | f.data = release.NewDirStore(f.root) |
| 33 | f.fx.srv.cfg.SeedData = f.data |
| 34 | enableInstances(t, f.fx, fromApp, fromManifest(true)) |
| 35 | leases := leasetest.NewVersionedBucket() |
| 36 | leases.AddBucket(f.fx.srv.cfg.HibernateBucket) |
| 37 | handler, _ := leasetest.NewS3Server(leases) |
| 38 | srv := httptest.NewServer(handler) |
| 39 | t.Cleanup(srv.Close) |
| 40 | t.Setenv("AWS_ACCESS_KEY_ID", "test-key") |
| 41 | t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret") |
| 42 | t.Setenv("AWS_SESSION_TOKEN", "") |
| 43 | t.Setenv("AWS_EC2_METADATA_DISABLED", "true") |
| 44 | t.Setenv("AWS_CONFIG_FILE", t.TempDir()+"/absent") |
| 45 | t.Setenv("AWS_SHARED_CREDENTIALS_FILE", t.TempDir()+"/absent") |
| 46 | t.Setenv("CARLOS_S3_ENDPOINT", srv.URL) |
| 47 | f.fx.srv.cfg.HibernateRegion = "test-region" |
| 48 | finishLeases[f.fx.srv] = leases |
| 49 | t.Cleanup(func() { delete(finishLeases, f.fx.srv) }) |
| 50 | return f |
| 51 | } |
| 52 | |
| 53 | // finishLeases is each finish fixture's lease bucket. |
| 54 | var finishLeases = map[*Server]*leasetest.VersionedBucket{} |
| 55 | |
| 56 | // putLease writes host's lock.json as the litestream leaser does, expiring |
| 57 | // at expires. |
| 58 | func (f fromFixture) putLease(t *testing.T, host string, expires time.Time) { |
| 59 | t.Helper() |
| 60 | finishLeases[f.fx.srv].Put(f.fx.srv.cfg.HibernateBucket, "instances/"+host+"/lock.json", |
| 61 | []byte(`{"generation":1,"expires_at":"`+expires.UTC().Format(time.RFC3339Nano)+`","owner":"box-a"}`)) |
| 62 | } |
| 63 | |
| 64 | // putReplicaKeys writes n stand-in LTX objects under host's replica prefix. |
| 65 | func (f fromFixture) putReplicaKeys(t *testing.T, host string, n int) []string { |
| 66 | t.Helper() |
| 67 | var keys []string |
| 68 | for i := 1; i <= n; i++ { |
| 69 | k := fmt.Sprintf("instances/%s/0000/%016x-%016x.ltx", host, i, i) |
| 70 | if _, err := f.data.Put(context.Background(), k, []byte("x"), ""); err != nil { |
| 71 | t.Fatal(err) |
| 72 | } |
| 73 | keys = append(keys, k) |
| 74 | } |
| 75 | return keys |
| 76 | } |
| 77 | |
| 78 | // rollBack walks a seed the way a box does once it has been aborted: the |
| 79 | // abort, then the box's CAS to rolled-back. |
| 80 | func (f fromFixture) rollBack(t *testing.T, host, id string) { |
| 81 | t.Helper() |
| 82 | ctx := context.Background() |
| 83 | for _, step := range []func(*provision.Record) error{provision.SeedAbort(id), provision.SeedRolledBack(id)} { |
| 84 | if _, err := provision.SetRecordSeed(ctx, f.fx.rel, "bab", fromApp, host, step); err != nil { |
| 85 | t.Fatal(err) |
| 86 | } |
| 87 | } |
| 88 | } |
| 89 | |
| 90 | // seedRecord writes a record of shop on host in env whose seed is id, of |
| 91 | // kind, in state. |
| 92 | func (f fromFixture) seedRecord(t *testing.T, host, env, id, kind, state string) { |
| 93 | t.Helper() |
| 94 | rec := provision.Record{Schema: 1, Host: host, Slug: "canary", App: fromApp, Account: "bab", Channel: "stable", |
| 95 | Environment: env, CreatedAt: time.Now().UTC(), |
| 96 | Seed: &provision.SeedSource{ID: id, Kind: kind, State: state, By: "bob@keymail.dev", At: time.Now().UTC()}} |
| 97 | if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil { |
| 98 | t.Fatal(err) |
| 99 | } |
| 100 | } |
| 101 | |
| 102 | // markedStore writes a canary store holding a grant, made by seed ("" |
| 103 | // for a store no seed made). |
| 104 | func (f fromFixture) markedStore(t *testing.T, seed string) { |
| 105 | t.Helper() |
| 106 | at := time.Now().UTC().Add(-time.Hour) |
| 107 | putCanaryStoreOf(t, f.fx, fromApp, &objstore.Record{ |
| 108 | Declared: &objstore.Declared{EnvPrefix: "S3", At: at, By: "bob@keymail.dev", Seed: seed}, |
| 109 | Grant: &objstore.GrantBlock{Backend: "mem", Bucket: "carlos-bab-shop-canary", Source: objstore.GrantSourceAccountApproval, At: at, By: "operator@keymail.dev"}, |
| 110 | }) |
| 111 | } |
| 112 | |
| 113 | func (f fromFixture) canaryStore(t *testing.T) *objstore.Record { |
| 114 | t.Helper() |
| 115 | rec, _, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "canary") |
| 116 | if err != nil { |
| 117 | t.Fatal(err) |
| 118 | } |
| 119 | return rec |
| 120 | } |
| 121 | |
| 122 | func (f fromFixture) rawKey(t *testing.T, s release.Store, key string) string { |
| 123 | t.Helper() |
| 124 | b, _, err := s.Get(context.Background(), key) |
| 125 | if err != nil { |
| 126 | t.Fatalf("read %s: %v", key, err) |
| 127 | } |
| 128 | return string(b) |
| 129 | } |
| 130 | |
| 131 | func (f fromFixture) requireRecordGone(t *testing.T, host string) { |
| 132 | t.Helper() |
| 133 | if rec, err := f.record(t, host); !errors.Is(err, os.ErrNotExist) { |
| 134 | t.Fatalf("record %s after the finish = %+v, %v; want it gone", host, rec, err) |
| 135 | } |
| 136 | } |
| 137 | |
| 138 | func newSeedID(t *testing.T) string { |
| 139 | t.Helper() |
| 140 | id, err := seeds.NewID() |
| 141 | if err != nil { |
| 142 | t.Fatal(err) |
| 143 | } |
| 144 | return id |
| 145 | } |
| 146 | |
| 147 | func replicaKeys(t *testing.T, f fromFixture, host string) []string { |
| 148 | t.Helper() |
| 149 | keys, err := f.data.List(context.Background(), "instances/"+host+"/") |
| 150 | if err != nil { |
| 151 | t.Fatal(err) |
| 152 | } |
| 153 | return keys |
| 154 | } |
| 155 | |
| 156 | // The whole finish for a --from seed that failed after its first upload: |
| 157 | // its own store torn down for real, its snapshot and its replica deleted, |
| 158 | // and the record last. The host is then free for the same copy again: no |
| 159 | // leftover store for §4 to refuse, no replica for the #199 rail to refuse. |
| 160 | func TestFinishSeedTearsDownAndFreesTheHost(t *testing.T) { |
| 161 | f := finishFixture(t) |
| 162 | rec := f.copy(t, fromBody) |
| 163 | requireReady(t, rec) |
| 164 | id := rec.Seed.ID |
| 165 | store := f.canaryStore(t) |
| 166 | if store == nil || store.Declared == nil || store.Declared.Seed != id { |
| 167 | t.Fatalf("canary store = %+v; want the copy's own", store) |
| 168 | } |
| 169 | bucket := storeBucketName(store) |
| 170 | if _, ok := f.fake.Buckets[bucket]; !ok { |
| 171 | t.Fatalf("bucket %s was never made", bucket) |
| 172 | } |
| 173 | // A box placed the copy and uploaded its first replica before the |
| 174 | // abort landed (a lost publish answer, Plan 5 T5). |
| 175 | dbdoortest.Replicate(t, filepath.Join(f.root, "instances", fromHost), 2) |
| 176 | if held, err := f.fx.srv.hostHoldsReplicaData(context.Background(), fromHost); err != nil || !held { |
| 177 | t.Fatalf("replica before the finish: %v, %v", held, err) |
| 178 | } |
| 179 | sourceReplica := replicaKeys(t, f, fromSource) |
| 180 | f.rollBack(t, fromHost, id) |
| 181 | |
| 182 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 183 | t.Fatal(err) |
| 184 | } |
| 185 | if got := f.canaryStore(t); got != nil { |
| 186 | t.Fatalf("store record after the finish = %+v; want it gone", got) |
| 187 | } |
| 188 | if _, ok := f.fake.Buckets[bucket]; ok { |
| 189 | t.Fatalf("bucket %s survived the teardown", bucket) |
| 190 | } |
| 191 | if _, _, err := f.data.Get(context.Background(), seeds.SnapshotKey(fromHost, id)); !errors.Is(err, os.ErrNotExist) { |
| 192 | t.Fatalf("snapshot after the finish: %v; want it gone", err) |
| 193 | } |
| 194 | if keys := replicaKeys(t, f, fromHost); len(keys) != 0 { |
| 195 | t.Fatalf("replica after the finish: %v", keys) |
| 196 | } |
| 197 | if held, err := f.fx.srv.hostHoldsReplicaData(context.Background(), fromHost); err != nil || held { |
| 198 | t.Fatalf("the rail after the finish: %v, %v", held, err) |
| 199 | } |
| 200 | if got := replicaKeys(t, f, fromSource); !slices.Equal(got, sourceReplica) { |
| 201 | t.Fatalf("the source's replica changed:\n before %v\n after %v", sourceReplica, got) |
| 202 | } |
| 203 | f.requireRecordGone(t, fromHost) |
| 204 | |
| 205 | // Idempotent: a second finish (the sweep after an inline one) is done. |
| 206 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 207 | t.Fatalf("a second finish: %v", err) |
| 208 | } |
| 209 | again := f.copy(t, fromBody) |
| 210 | requireReady(t, again) |
| 211 | if again.Seed.ID == id { |
| 212 | t.Fatal("the new copy reused the old seed id") |
| 213 | } |
| 214 | if store := f.canaryStore(t); store == nil || store.Declared == nil || store.Declared.Seed != again.Seed.ID { |
| 215 | t.Fatalf("the new copy's store = %+v; want one marked with %s", store, again.Seed.ID) |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // A teardown that fails leaves the record rolled-back and holding the host, |
| 220 | // so the next finish (the sweep's) resumes and completes. |
| 221 | func TestFinishSeedResumesAfterAFailedTeardown(t *testing.T) { |
| 222 | f := finishFixture(t) |
| 223 | rec := f.copy(t, fromBody) |
| 224 | requireReady(t, rec) |
| 225 | id := rec.Seed.ID |
| 226 | f.rollBack(t, fromHost, id) |
| 227 | f.fake.FailNext["DeleteBucket"] = errors.New("the backend is down") |
| 228 | |
| 229 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 230 | t.Fatal("a failed teardown answered nil") |
| 231 | } |
| 232 | if got, err := f.record(t, fromHost); err != nil || got.Seed == nil || got.Seed.State != seeds.StateRolledBack { |
| 233 | t.Fatalf("record after a failed teardown = %+v, %v; want it kept, rolled-back", got, err) |
| 234 | } |
| 235 | if f.canaryStore(t) == nil { |
| 236 | t.Fatal("the store record went before its estate") |
| 237 | } |
| 238 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 239 | t.Fatalf("the resumed finish: %v", err) |
| 240 | } |
| 241 | if got := f.canaryStore(t); got != nil { |
| 242 | t.Fatalf("store record after the resumed finish = %+v", got) |
| 243 | } |
| 244 | f.requireRecordGone(t, fromHost) |
| 245 | } |
| 246 | |
| 247 | // Only a store this seed made is torn down. Any other store is left as it |
| 248 | // was, and the record is still deleted. |
| 249 | func TestFinishSeedLeavesAStoreItDidNotMake(t *testing.T) { |
| 250 | for name, seed := range map[string]string{ |
| 251 | "another seed's": "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb", |
| 252 | "no seed's": "", |
| 253 | } { |
| 254 | t.Run(name, func(t *testing.T) { |
| 255 | f := finishFixture(t) |
| 256 | id := newSeedID(t) |
| 257 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 258 | f.markedStore(t, seed) |
| 259 | key := objstore.KeyIn("bab", fromApp, "canary") |
| 260 | before := f.rawKey(t, f.fx.rel, key) |
| 261 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 262 | t.Fatal(err) |
| 263 | } |
| 264 | if after := f.rawKey(t, f.fx.rel, key); after != before { |
| 265 | t.Fatalf("the store record changed:\n before %s\n after %s", before, after) |
| 266 | } |
| 267 | f.requireRecordGone(t, fromHost) |
| 268 | }) |
| 269 | } |
| 270 | } |
| 271 | |
| 272 | // A record the finish does not own is left alone, with the store its seed |
| 273 | // made: another seed, a seed not yet rolled back, no seed at all. |
| 274 | func TestFinishSeedLeavesARecordItDoesNotOwn(t *testing.T) { |
| 275 | id := "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa" |
| 276 | for name, c := range map[string]struct{ id, state string }{ |
| 277 | "another seed": {"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb", seeds.StateRolledBack}, |
| 278 | "aborted, not yet": {id, seeds.StateAborted}, |
| 279 | "ready": {id, seeds.StateReady}, |
| 280 | "placed": {id, seeds.StatePlaced}, |
| 281 | "seedless (no seed)": {"", ""}, |
| 282 | } { |
| 283 | t.Run(name, func(t *testing.T) { |
| 284 | f := finishFixture(t) |
| 285 | if c.id == "" { |
| 286 | if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: fromHost, Slug: "canary", |
| 287 | App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil { |
| 288 | t.Fatal(err) |
| 289 | } |
| 290 | } else { |
| 291 | f.seedRecord(t, fromHost, "canary", c.id, seeds.KindReplica, c.state) |
| 292 | } |
| 293 | f.markedStore(t, id) |
| 294 | if _, err := f.data.Put(context.Background(), "instances/"+fromHost+"/0000/0000000000000001-0000000000000001.ltx", []byte("x"), ""); err != nil { |
| 295 | t.Fatal(err) |
| 296 | } |
| 297 | recKey := provision.RecordKey("bab", fromApp, fromHost) |
| 298 | storeKey := objstore.KeyIn("bab", fromApp, "canary") |
| 299 | recBefore, storeBefore := f.rawKey(t, f.fx.rel, recKey), f.rawKey(t, f.fx.rel, storeKey) |
| 300 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 301 | t.Fatal(err) |
| 302 | } |
| 303 | if after := f.rawKey(t, f.fx.rel, recKey); after != recBefore { |
| 304 | t.Fatalf("the record changed:\n before %s\n after %s", recBefore, after) |
| 305 | } |
| 306 | if after := f.rawKey(t, f.fx.rel, storeKey); after != storeBefore { |
| 307 | t.Fatalf("the store record changed:\n before %s\n after %s", storeBefore, after) |
| 308 | } |
| 309 | if keys := replicaKeys(t, f, fromHost); len(keys) != 1 { |
| 310 | t.Fatalf("replica = %v; want it untouched", keys) |
| 311 | } |
| 312 | }) |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // The finish decides under the store's walk lock (preflight M9). A grant |
| 317 | // by another writer that held the lock while the finish queued, and took |
| 318 | // the store over (clearing its seed mark), wins: the finish re-reads under |
| 319 | // the lock and leaves the store standing. |
| 320 | func TestFinishSeedReReadsTheStoreUnderItsLock(t *testing.T) { |
| 321 | f := finishFixture(t) |
| 322 | id := newSeedID(t) |
| 323 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 324 | f.markedStore(t, id) |
| 325 | unlock := f.fx.srv.storeWalks.lock("bab", fromApp, "canary") |
| 326 | done := make(chan error, 1) |
| 327 | go func() { done <- f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id) }() |
| 328 | deadline := time.Now().Add(3 * time.Second) |
| 329 | for { |
| 330 | f.fx.srv.storeWalks.mu.Lock() |
| 331 | refs := 0 |
| 332 | if e := f.fx.srv.storeWalks.m[storeWalkKey("bab", fromApp, "canary")]; e != nil { |
| 333 | refs = e.refs |
| 334 | } |
| 335 | f.fx.srv.storeWalks.mu.Unlock() |
| 336 | if refs == 2 { |
| 337 | break |
| 338 | } |
| 339 | if time.Now().After(deadline) { |
| 340 | unlock() |
| 341 | t.Fatalf("the finish never queued on the store lock: refs=%d", refs) |
| 342 | } |
| 343 | time.Sleep(time.Millisecond) |
| 344 | } |
| 345 | store, etag, err := objstore.GetIn(context.Background(), f.fx.rel, "bab", fromApp, "canary") |
| 346 | if err != nil { |
| 347 | t.Fatal(err) |
| 348 | } |
| 349 | store.Declared.Seed = "" |
| 350 | if _, err := objstore.PutIn(context.Background(), f.fx.rel, "bab", fromApp, "canary", store, etag); err != nil { |
| 351 | t.Fatal(err) |
| 352 | } |
| 353 | key := objstore.KeyIn("bab", fromApp, "canary") |
| 354 | taken := f.rawKey(t, f.fx.rel, key) |
| 355 | unlock() |
| 356 | if err := <-done; err != nil { |
| 357 | t.Fatal(err) |
| 358 | } |
| 359 | if after := f.rawKey(t, f.fx.rel, key); after != taken { |
| 360 | t.Fatalf("a store taken over under the lock was touched:\n before %s\n after %s", taken, after) |
| 361 | } |
| 362 | f.requireRecordGone(t, fromHost) |
| 363 | } |
| 364 | |
| 365 | // Defence in depth (preflight M10): a store this seed made is not torn down |
| 366 | // while another instance names its environment. The record still goes: |
| 367 | // the store now serves the other instance, and an operator decides. |
| 368 | func TestFinishSeedSparesAStoreAnotherInstanceUses(t *testing.T) { |
| 369 | f := finishFixture(t) |
| 370 | id := newSeedID(t) |
| 371 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 372 | if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: "other.shop.example.com", Slug: "other", |
| 373 | App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil { |
| 374 | t.Fatal(err) |
| 375 | } |
| 376 | f.markedStore(t, id) |
| 377 | key := objstore.KeyIn("bab", fromApp, "canary") |
| 378 | before := f.rawKey(t, f.fx.rel, key) |
| 379 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 380 | t.Fatal(err) |
| 381 | } |
| 382 | if after := f.rawKey(t, f.fx.rel, key); after != before { |
| 383 | t.Fatalf("a shared store was touched:\n before %s\n after %s", before, after) |
| 384 | } |
| 385 | f.requireRecordGone(t, fromHost) |
| 386 | } |
| 387 | |
| 388 | // A record of the app that will not decode may name the environment: the |
| 389 | // finish cannot know it does not, so the store is spared as for a sharer. |
| 390 | // The sweep runs this finish unattended (Plan 8 M7). |
| 391 | func TestFinishSeedSparesAStoreWhenARecordIsUnreadable(t *testing.T) { |
| 392 | f := finishFixture(t) |
| 393 | id := newSeedID(t) |
| 394 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 395 | if _, err := f.fx.rel.Put(context.Background(), provision.RecordKey("bab", fromApp, "other.shop.example.com"), []byte("{not json"), ""); err != nil { |
| 396 | t.Fatal(err) |
| 397 | } |
| 398 | f.markedStore(t, id) |
| 399 | key := objstore.KeyIn("bab", fromApp, "canary") |
| 400 | before := f.rawKey(t, f.fx.rel, key) |
| 401 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 402 | t.Fatal(err) |
| 403 | } |
| 404 | if after := f.rawKey(t, f.fx.rel, key); after != before { |
| 405 | t.Fatalf("a store an unreadable record may use was touched:\n before %s\n after %s", before, after) |
| 406 | } |
| 407 | } |
| 408 | |
| 409 | // failingRecordList fails every listing of the app's records. |
| 410 | type failingRecordList struct { |
| 411 | release.Store |
| 412 | prefix string |
| 413 | } |
| 414 | |
| 415 | func (s failingRecordList) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| 416 | return release.DeleteIfMatch(ctx, s.Store, key, etag) |
| 417 | } |
| 418 | |
| 419 | func (s failingRecordList) List(ctx context.Context, prefix string) ([]string, error) { |
| 420 | if prefix == s.prefix { |
| 421 | return nil, errors.New("the bucket is unreachable") |
| 422 | } |
| 423 | return s.Store.List(ctx, prefix) |
| 424 | } |
| 425 | |
| 426 | // When the records cannot be listed the finish cannot tell whether the |
| 427 | // store is shared, so it does nothing and answers an error for the sweep |
| 428 | // to retry. |
| 429 | func TestFinishSeedRetriesWhenTheRecordsCannotBeListed(t *testing.T) { |
| 430 | f := finishFixture(t) |
| 431 | id := newSeedID(t) |
| 432 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 433 | f.markedStore(t, id) |
| 434 | rel := f.fx.srv.cfg.Releases |
| 435 | f.fx.srv.cfg.Releases = failingRecordList{Store: rel, prefix: provision.RecordsPrefix("bab", fromApp)} |
| 436 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 437 | t.Fatal("a failed listing answered nil") |
| 438 | } |
| 439 | if f.canaryStore(t) == nil { |
| 440 | t.Fatal("the store record went on a failed listing") |
| 441 | } |
| 442 | if _, err := f.record(t, fromHost); err != nil { |
| 443 | t.Fatalf("the record went on a failed listing: %v", err) |
| 444 | } |
| 445 | } |
| 446 | |
| 447 | // A store this seed made with no backend to tear it down with is an error, |
| 448 | // and the record keeps holding the host. |
| 449 | func TestFinishSeedNeedsAStorageBackend(t *testing.T) { |
| 450 | f := finishFixture(t) |
| 451 | id := newSeedID(t) |
| 452 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 453 | f.markedStore(t, id) |
| 454 | f.fx.srv.cfg.StoreProvisioner = nil |
| 455 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 456 | t.Fatal("no backend answered nil") |
| 457 | } |
| 458 | if _, err := f.record(t, fromHost); err != nil { |
| 459 | t.Fatalf("the record went with its store still standing: %v", err) |
| 460 | } |
| 461 | } |
| 462 | |
| 463 | // An import's finish deletes its snapshot and never the host's replica: |
| 464 | // only a --from seed's replica is known to be its own. |
| 465 | func TestFinishSeedLeavesAnImportsReplica(t *testing.T) { |
| 466 | f := finishFixture(t) |
| 467 | id := newSeedID(t) |
| 468 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 469 | ctx := context.Background() |
| 470 | if _, err := f.data.Put(ctx, seeds.SnapshotKey(fromHost, id), []byte("snapshot"), ""); err != nil { |
| 471 | t.Fatal(err) |
| 472 | } |
| 473 | dbdoortest.Replicate(t, filepath.Join(f.root, "instances", fromHost), 2) |
| 474 | before := replicaKeys(t, f, fromHost) |
| 475 | if err := f.fx.srv.finishSeed(ctx, "bab", fromApp, fromHost, id); err != nil { |
| 476 | t.Fatal(err) |
| 477 | } |
| 478 | if _, _, err := f.data.Get(ctx, seeds.SnapshotKey(fromHost, id)); !errors.Is(err, os.ErrNotExist) { |
| 479 | t.Fatalf("snapshot after the finish: %v; want it gone", err) |
| 480 | } |
| 481 | if after := replicaKeys(t, f, fromHost); len(before) == 0 || !slices.Equal(after, before) { |
| 482 | t.Fatalf("an import's finish touched the replica:\n before %v\n after %v", before, after) |
| 483 | } |
| 484 | f.requireRecordGone(t, fromHost) |
| 485 | } |
| 486 | |
| 487 | // A seed id that fails seeds.Check builds no key and deletes nothing |
| 488 | // (preflight M8): every bad id would map to the one shared sentinel. |
| 489 | func TestFinishSeedChecksTheSeedBeforeAnyKey(t *testing.T) { |
| 490 | f := finishFixture(t) |
| 491 | ctx := context.Background() |
| 492 | sentinel := seeds.SnapshotKey(fromHost, "sd_bad") |
| 493 | if _, err := f.data.Put(ctx, sentinel, []byte("someone else's"), ""); err != nil { |
| 494 | t.Fatal(err) |
| 495 | } |
| 496 | // A rolled-back record carrying the bad id: without the check, the |
| 497 | // finish would match it and delete the sentinel. |
| 498 | f.seedRecord(t, fromHost, "canary", "sd_bad", seeds.KindImport, seeds.StateRolledBack) |
| 499 | for _, c := range []struct{ host, id string }{{fromHost, "sd_bad"}, {"../shop", newSeedID(t)}} { |
| 500 | if err := f.fx.srv.finishSeed(ctx, "bab", fromApp, c.host, c.id); !errors.Is(err, seeds.ErrBadID) && !errors.Is(err, seeds.ErrBadHost) { |
| 501 | t.Fatalf("finishSeed(%q, %q) = %v; want a seeds.Check error", c.host, c.id, err) |
| 502 | } |
| 503 | } |
| 504 | if _, _, err := f.data.Get(ctx, sentinel); err != nil { |
| 505 | t.Fatalf("the sentinel key was deleted: %v", err) |
| 506 | } |
| 507 | } |
| 508 | |
| 509 | // No record: nothing to finish, and no error. |
| 510 | func TestFinishSeedWithNoRecord(t *testing.T) { |
| 511 | f := finishFixture(t) |
| 512 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, newSeedID(t)); err != nil { |
| 513 | t.Fatal(err) |
| 514 | } |
| 515 | } |
| 516 | |
| 517 | // rolledBackFrom writes a --from record rolled back on host, with replica |
| 518 | // keys under its prefix, and answers its seed id and those keys. |
| 519 | func (f fromFixture) rolledBackFrom(t *testing.T, host string) (string, []string) { |
| 520 | t.Helper() |
| 521 | id := newSeedID(t) |
| 522 | f.seedRecord(t, host, "canary", id, seeds.KindReplica, seeds.StateRolledBack) |
| 523 | return id, f.putReplicaKeys(t, host, 3) |
| 524 | } |
| 525 | |
| 526 | // Review I1 (b): a lease that is present and unexpired by the lease |
| 527 | // package's own rule blocks the replica step. Nothing is deleted, the |
| 528 | // error is retryable and the record stays rolled-back. An absent lease, or |
| 529 | // one expired past its margin (a box that died), does not block. |
| 530 | func TestFinishSeedWaitsForTheHostsLease(t *testing.T) { |
| 531 | // Expiries are relative to the moment each case writes its lease: the |
| 532 | // margin is seconds, and earlier cases take most of one. |
| 533 | for name, c := range map[string]struct { |
| 534 | expiresIn *time.Duration |
| 535 | blocked bool |
| 536 | }{ |
| 537 | "held": {expiresIn: ptr(time.Minute), blocked: true}, |
| 538 | "just expired": {expiresIn: ptr(-lease.RenewMargin / 2), blocked: true}, |
| 539 | "expired past margin": {expiresIn: ptr(-2 * lease.RenewMargin)}, |
| 540 | "absent": {}, |
| 541 | } { |
| 542 | t.Run(name, func(t *testing.T) { |
| 543 | f := finishFixture(t) |
| 544 | id, keys := f.rolledBackFrom(t, fromHost) |
| 545 | if c.expiresIn != nil { |
| 546 | f.putLease(t, fromHost, time.Now().Add(*c.expiresIn)) |
| 547 | } |
| 548 | err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id) |
| 549 | if !c.blocked { |
| 550 | if err != nil { |
| 551 | t.Fatal(err) |
| 552 | } |
| 553 | if got := replicaKeys(t, f, fromHost); len(got) != 0 { |
| 554 | t.Fatalf("replica after the finish: %v", got) |
| 555 | } |
| 556 | f.requireRecordGone(t, fromHost) |
| 557 | return |
| 558 | } |
| 559 | if err == nil { |
| 560 | t.Fatal("a held lease answered nil") |
| 561 | } |
| 562 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 563 | t.Fatalf("replica under a held lease:\n before %v\n after %v", keys, got) |
| 564 | } |
| 565 | if got, err := f.record(t, fromHost); err != nil || got.Seed.State != seeds.StateRolledBack { |
| 566 | t.Fatalf("record under a held lease = %+v, %v; want it kept, rolled-back", got, err) |
| 567 | } |
| 568 | }) |
| 569 | } |
| 570 | } |
| 571 | |
| 572 | func ptr[T any](v T) *T { return &v } |
| 573 | |
| 574 | // recreatingData runs recreate once: before the first Delete whose key |
| 575 | // onDelete matches, or after the first List whose prefix onList matches. |
| 576 | type recreatingData struct { |
| 577 | release.Store |
| 578 | onDelete, onList func(key string) bool |
| 579 | recreate func() |
| 580 | done bool |
| 581 | } |
| 582 | |
| 583 | func (d *recreatingData) fire(match func(string) bool, key string) { |
| 584 | if !d.done && match != nil && match(key) { |
| 585 | d.done = true |
| 586 | d.recreate() |
| 587 | } |
| 588 | } |
| 589 | |
| 590 | func (d *recreatingData) Delete(ctx context.Context, key string) error { |
| 591 | d.fire(d.onDelete, key) |
| 592 | return d.Store.Delete(ctx, key) |
| 593 | } |
| 594 | |
| 595 | func (d *recreatingData) List(ctx context.Context, prefix string) ([]string, error) { |
| 596 | keys, err := d.Store.List(ctx, prefix) |
| 597 | d.fire(d.onList, prefix) |
| 598 | return keys, err |
| 599 | } |
| 600 | |
| 601 | // recreateHost deletes host's record and writes a new, seedless instance |
| 602 | // of it, as another finish and then a new create would. |
| 603 | func (f fromFixture) recreateHost(t *testing.T, host string) func() { |
| 604 | return func() { |
| 605 | ctx := context.Background() |
| 606 | if err := provision.DeleteRecord(ctx, f.fx.rel, "bab", fromApp, host); err != nil { |
| 607 | t.Error(err) |
| 608 | } |
| 609 | if err := provision.CreateRecord(ctx, f.fx.rel, provision.Record{Schema: 1, Host: host, Slug: "canary", |
| 610 | App: fromApp, Account: "bab", Channel: "stable", Environment: "canary", CreatedAt: time.Now().UTC()}); err != nil { |
| 611 | t.Error(err) |
| 612 | } |
| 613 | } |
| 614 | } |
| 615 | |
| 616 | // Review I2: the replica delete is fenced on the record. A record deleted |
| 617 | // and created again for the same host, between the store teardown and the |
| 618 | // replica step, or while the replica is listed, keeps every replica key: |
| 619 | // they may be the new instance's. The finish answers an error (the |
| 620 | // record it was finishing is not gone in the sense that matters: the host |
| 621 | // has a record), and the new record is untouched. |
| 622 | func TestFinishSeedNeverDeletesANewInstancesReplica(t *testing.T) { |
| 623 | for name, wrap := range map[string]func(f fromFixture) *recreatingData{ |
| 624 | "before the replica step": func(f fromFixture) *recreatingData { |
| 625 | return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost), |
| 626 | onDelete: func(key string) bool { return strings.HasPrefix(key, seeds.Prefix) }} |
| 627 | }, |
| 628 | "while it is listed": func(f fromFixture) *recreatingData { |
| 629 | return &recreatingData{Store: f.data, recreate: f.recreateHost(t, fromHost), |
| 630 | onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") }} |
| 631 | }, |
| 632 | } { |
| 633 | t.Run(name, func(t *testing.T) { |
| 634 | f := finishFixture(t) |
| 635 | id, keys := f.rolledBackFrom(t, fromHost) |
| 636 | d := wrap(f) |
| 637 | f.fx.srv.cfg.SeedData = d |
| 638 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 639 | t.Fatal("a re-created host answered nil") |
| 640 | } |
| 641 | if !d.done { |
| 642 | t.Fatal("the record was never re-created: the test did not reach its window") |
| 643 | } |
| 644 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 645 | t.Fatalf("the new instance's replica:\n before %v\n after %v", keys, got) |
| 646 | } |
| 647 | if got, err := f.record(t, fromHost); err != nil || got.Seed != nil { |
| 648 | t.Fatalf("the new record = %+v, %v; want it untouched", got, err) |
| 649 | } |
| 650 | // The sweep's next finish of the old seed leaves the new |
| 651 | // instance alone and is done. |
| 652 | f.fx.srv.cfg.SeedData = f.data |
| 653 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 654 | t.Fatal(err) |
| 655 | } |
| 656 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 657 | t.Fatalf("the retry touched the new instance's replica: %v", got) |
| 658 | } |
| 659 | }) |
| 660 | } |
| 661 | } |
| 662 | |
| 663 | // A record rewritten in place while the replica is listed (an ordinary |
| 664 | // writer, the same seed still rolled-back) stops the deletes too: the |
| 665 | // finish only deletes under the record it read. The next finish resumes. |
| 666 | func TestFinishSeedStopsWhenTheRecordChangesMidReplica(t *testing.T) { |
| 667 | f := finishFixture(t) |
| 668 | id, keys := f.rolledBackFrom(t, fromHost) |
| 669 | key := provision.RecordKey("bab", fromApp, fromHost) |
| 670 | f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data, |
| 671 | onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") }, |
| 672 | recreate: func() { |
| 673 | ctx := context.Background() |
| 674 | raw, etag, err := f.fx.rel.Get(ctx, key) |
| 675 | if err == nil { |
| 676 | _, err = f.fx.rel.Put(ctx, key, append(raw, '\n'), etag) |
| 677 | } |
| 678 | if err != nil { |
| 679 | t.Error(err) |
| 680 | } |
| 681 | }} |
| 682 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 683 | t.Fatal("a record changed mid-replica answered nil") |
| 684 | } |
| 685 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 686 | t.Fatalf("replica after the record changed:\n before %v\n after %v", keys, got) |
| 687 | } |
| 688 | f.fx.srv.cfg.SeedData = f.data |
| 689 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 690 | t.Fatal(err) |
| 691 | } |
| 692 | if got := replicaKeys(t, f, fromHost); len(got) != 0 { |
| 693 | t.Fatalf("replica after the resumed finish: %v", got) |
| 694 | } |
| 695 | f.requireRecordGone(t, fromHost) |
| 696 | } |
| 697 | |
| 698 | // A record deleted (not re-created) during the replica step was finished |
| 699 | // by another finish: nil, and nothing more is deleted. |
| 700 | func TestFinishSeedStopsWhenAnotherFinishDeletedTheRecord(t *testing.T) { |
| 701 | f := finishFixture(t) |
| 702 | id, keys := f.rolledBackFrom(t, fromHost) |
| 703 | f.fx.srv.cfg.SeedData = &recreatingData{Store: f.data, |
| 704 | onList: func(prefix string) bool { return strings.HasPrefix(prefix, "instances/") }, |
| 705 | recreate: func() { |
| 706 | if err := provision.DeleteRecord(context.Background(), f.fx.rel, "bab", fromApp, fromHost); err != nil { |
| 707 | t.Error(err) |
| 708 | } |
| 709 | }} |
| 710 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 711 | t.Fatal(err) |
| 712 | } |
| 713 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 714 | t.Fatalf("replica after the record went: %v", got) |
| 715 | } |
| 716 | } |
| 717 | |
| 718 | // The prefix stops at the host: a host that extends this one's name keeps |
| 719 | // its replica. |
| 720 | func TestFinishSeedReplicaPrefixStopsAtTheHost(t *testing.T) { |
| 721 | f := finishFixture(t) |
| 722 | id, _ := f.rolledBackFrom(t, fromHost) |
| 723 | sibling := f.putReplicaKeys(t, fromHost+".au", 2) |
| 724 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 725 | t.Fatal(err) |
| 726 | } |
| 727 | if got := replicaKeys(t, f, fromHost); len(got) != 0 { |
| 728 | t.Fatalf("replica after the finish: %v", got) |
| 729 | } |
| 730 | if got := replicaKeys(t, f, fromHost+".au"); !slices.Equal(got, sibling) { |
| 731 | t.Fatalf("the sibling host's replica:\n before %v\n after %v", sibling, got) |
| 732 | } |
| 733 | } |
| 734 | |
| 735 | // A replica outside this console's data bucket (another region) is not |
| 736 | // known to be the seed's, and is left alone; the record still goes. |
| 737 | func TestFinishSeedLeavesAReplicaOutsideItsBucket(t *testing.T) { |
| 738 | f := finishFixture(t) |
| 739 | id := newSeedID(t) |
| 740 | rec := provision.Record{Schema: 1, Host: fromHost, Slug: "canary", App: fromApp, Account: "bab", Channel: "stable", |
| 741 | Environment: "canary", Region: "elsewhere", CreatedAt: time.Now().UTC(), |
| 742 | Seed: &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateRolledBack, By: "bob@keymail.dev", At: time.Now().UTC()}} |
| 743 | if err := provision.CreateRecord(context.Background(), f.fx.rel, rec); err != nil { |
| 744 | t.Fatal(err) |
| 745 | } |
| 746 | keys := f.putReplicaKeys(t, fromHost, 2) |
| 747 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 748 | t.Fatal(err) |
| 749 | } |
| 750 | if got := replicaKeys(t, f, fromHost); !slices.Equal(got, keys) { |
| 751 | t.Fatalf("a replica outside this bucket was touched: %v", got) |
| 752 | } |
| 753 | f.requireRecordGone(t, fromHost) |
| 754 | } |
| 755 | |
| 756 | // More instances than one listing holds: the finish cannot tell whether |
| 757 | // the store is shared, so it does nothing and answers an error. |
| 758 | func TestFinishSeedRefusesATruncatedListing(t *testing.T) { |
| 759 | f := finishFixture(t) |
| 760 | id := newSeedID(t) |
| 761 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 762 | f.markedStore(t, id) |
| 763 | for i := range instanceListMax { |
| 764 | if err := provision.CreateRecord(context.Background(), f.fx.rel, provision.Record{Schema: 1, Host: fmt.Sprintf("i%d.shop.example.com", i), |
| 765 | Slug: fmt.Sprintf("i%d", i), App: fromApp, Account: "bab", Channel: "stable", CreatedAt: time.Now().UTC()}); err != nil { |
| 766 | t.Fatal(err) |
| 767 | } |
| 768 | } |
| 769 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 770 | t.Fatal("a truncated listing answered nil") |
| 771 | } |
| 772 | if f.canaryStore(t) == nil { |
| 773 | t.Fatal("the store went on a truncated listing") |
| 774 | } |
| 775 | if _, err := f.record(t, fromHost); err != nil { |
| 776 | t.Fatalf("the record went on a truncated listing: %v", err) |
| 777 | } |
| 778 | } |
| 779 | |
| 780 | // touchOnDelete rewrites the record once (an ordinary writer's update) |
| 781 | // just before the finish's conditional delete, so that delete loses. |
| 782 | type touchOnDelete struct { |
| 783 | release.Store |
| 784 | key string |
| 785 | touched bool |
| 786 | } |
| 787 | |
| 788 | func (s *touchOnDelete) DeleteIfMatch(ctx context.Context, key, etag string) error { |
| 789 | if key == s.key && !s.touched { |
| 790 | s.touched = true |
| 791 | raw, cur, err := s.Store.Get(ctx, key) |
| 792 | if err != nil { |
| 793 | return err |
| 794 | } |
| 795 | if _, err := s.Store.Put(ctx, key, append(raw, '\n'), cur); err != nil { |
| 796 | return err |
| 797 | } |
| 798 | } |
| 799 | return release.DeleteIfMatch(ctx, s.Store, key, etag) |
| 800 | } |
| 801 | |
| 802 | // A record changed between the finish's read and its delete is read again |
| 803 | // once, and the finish completes. |
| 804 | func TestFinishSeedRetriesALostRecordDelete(t *testing.T) { |
| 805 | f := finishFixture(t) |
| 806 | id := newSeedID(t) |
| 807 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 808 | rel := &touchOnDelete{Store: f.fx.rel, key: provision.RecordKey("bab", fromApp, fromHost)} |
| 809 | f.fx.srv.cfg.Releases = rel |
| 810 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err != nil { |
| 811 | t.Fatal(err) |
| 812 | } |
| 813 | if !rel.touched { |
| 814 | t.Fatal("the record was never touched: the test did not reach its window") |
| 815 | } |
| 816 | f.requireRecordGone(t, fromHost) |
| 817 | } |
| 818 | |
| 819 | // A console with no data bucket refuses before anything is destroyed: the |
| 820 | // store this seed made still stands, with the record. |
| 821 | func TestFinishSeedNeedsADataBucketBeforeTheTeardown(t *testing.T) { |
| 822 | f := finishFixture(t) |
| 823 | id := newSeedID(t) |
| 824 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 825 | f.markedStore(t, id) |
| 826 | f.fx.srv.cfg.SeedData = nil |
| 827 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 828 | t.Fatal("no data bucket answered nil") |
| 829 | } |
| 830 | if f.canaryStore(t) == nil { |
| 831 | t.Fatal("the store was torn down by a finish that could not complete") |
| 832 | } |
| 833 | if _, err := f.record(t, fromHost); err != nil { |
| 834 | t.Fatalf("the record went: %v", err) |
| 835 | } |
| 836 | } |
| 837 | |
| 838 | // A console with host claims armed finishes no seed. |
| 839 | func TestFinishSeedRefusesAClaimConsole(t *testing.T) { |
| 840 | f := finishFixture(t) |
| 841 | id := newSeedID(t) |
| 842 | f.seedRecord(t, fromHost, "canary", id, seeds.KindImport, seeds.StateRolledBack) |
| 843 | f.markedStore(t, id) |
| 844 | f.fx.srv.cfg.HostClaim = true |
| 845 | if err := f.fx.srv.finishSeed(context.Background(), "bab", fromApp, fromHost, id); err == nil { |
| 846 | t.Fatal("a claim console answered nil") |
| 847 | } |
| 848 | if f.canaryStore(t) == nil { |
| 849 | t.Fatal("a claim console tore the store down") |
| 850 | } |
| 851 | if _, err := f.record(t, fromHost); err != nil { |
| 852 | t.Fatalf("a claim console deleted the record: %v", err) |
| 853 | } |
| 854 | } |
| 855 | |
amadan