| 1 | package provision |
| 2 | |
| 3 | import ( |
| 4 | "bytes" |
| 5 | "context" |
| 6 | "database/sql" |
| 7 | "errors" |
| 8 | "log" |
| 9 | "os" |
| 10 | "path/filepath" |
| 11 | "reflect" |
| 12 | "strings" |
| 13 | "testing" |
| 14 | "time" |
| 15 | |
| 16 | "github.com/carlosframework/platform/internal/registry" |
| 17 | "github.com/carlosframework/platform/internal/release" |
| 18 | "github.com/carlosframework/platform/internal/seeds" |
| 19 | ) |
| 20 | |
| 21 | // captureLog collects the package's log lines for the rest of the test. |
| 22 | func captureLog(t *testing.T) *bytes.Buffer { |
| 23 | t.Helper() |
| 24 | var b bytes.Buffer |
| 25 | log.SetOutput(&b) |
| 26 | t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| 27 | return &b |
| 28 | } |
| 29 | |
| 30 | // journaled is an import this box has swapped: the gated exec row in |
| 31 | // place of the addr row, and the journal at swapped under the record's |
| 32 | // key. The record is whatever state was asked for. |
| 33 | type journaled struct { |
| 34 | s *release.DirStore |
| 35 | reg *registry.Registry |
| 36 | rec Record |
| 37 | row registry.Route |
| 38 | j registry.ImportJournal |
| 39 | cfg ReconcileConfig |
| 40 | } |
| 41 | |
| 42 | func journaledImport(t *testing.T, state string) journaled { |
| 43 | t.Helper() |
| 44 | s, sqid := shopSetup(t) |
| 45 | rec := newImport(sqid) |
| 46 | rec.Seed.State = state |
| 47 | if err := CreateRecord(ctx, s, rec); err != nil { |
| 48 | t.Fatal(err) |
| 49 | } |
| 50 | reg := regWith(t, addrRow(rec, sqid)) |
| 51 | original := mustGet(t, reg, rec.Host) |
| 52 | if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original, |
| 53 | RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, importRow(rec, "/data", nil)); err != nil { |
| 54 | t.Fatal(err) |
| 55 | } |
| 56 | j, err := reg.ImportJournal(rec.Host) |
| 57 | if err != nil || j == nil { |
| 58 | t.Fatalf("journal = %+v, %v", j, err) |
| 59 | } |
| 60 | return journaled{s: s, reg: reg, rec: rec, row: *mustGet(t, reg, rec.Host), j: *j, cfg: importCfg(t, nil)} |
| 61 | } |
| 62 | |
| 63 | // box wires the rollback's box hooks (seedrollback_test.go) into the |
| 64 | // scene, so a rollback shows by its effects and the lack of one by none. |
| 65 | func (x *journaled) box(t *testing.T) *rollbackBox { |
| 66 | t.Helper() |
| 67 | b := newRollbackBox(t, x.reg, x.j.Original) |
| 68 | b.hooks(x.cfg.Seeds) |
| 69 | x.cfg.Seeds.Data = faultData{x.cfg.Seeds.Data, b} |
| 70 | return b |
| 71 | } |
| 72 | |
| 73 | // rolledBack asserts the rollback ran to its end: the original route back |
| 74 | // exactly, the journal row gone, the host fenced before its local data |
| 75 | // went, and its replica deleted or kept as replicaGone says. |
| 76 | func (x journaled) rolledBack(t *testing.T, b *rollbackBox, replicaGone bool) { |
| 77 | t.Helper() |
| 78 | if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.j.Original) { |
| 79 | t.Fatalf("row = %+v, want the original %+v", *got, x.j.Original) |
| 80 | } |
| 81 | if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil { |
| 82 | t.Fatalf("journal = %+v", j) |
| 83 | } |
| 84 | if b.local || b.replica == replicaGone || count(b.steps, "fence") != 1 { |
| 85 | t.Fatalf("local %v replica %v steps %q", b.local, b.replica, b.steps) |
| 86 | } |
| 87 | } |
| 88 | |
| 89 | // untouched asserts the journal owned the host this pass: the swapped row |
| 90 | // and the journal row exactly as they were. |
| 91 | func (x journaled) untouched(t *testing.T) { |
| 92 | t.Helper() |
| 93 | if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.row) { |
| 94 | t.Fatalf("row changed:\n got %+v\nwant %+v", *got, x.row) |
| 95 | } |
| 96 | if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) { |
| 97 | t.Fatalf("journal changed: %+v", j) |
| 98 | } |
| 99 | } |
| 100 | |
| 101 | func TestJournalResumesTheRecordCAS(t *testing.T) { |
| 102 | s, sqid := shopSetup(t) |
| 103 | rec := importRecord(t, s, sqid) |
| 104 | reg := regWith(t, addrRow(rec, sqid)) |
| 105 | stored, _ := reg.Get(rec.Host) |
| 106 | // The box crashed after the transaction and before the record CAS. |
| 107 | next := RouteFor(rec, "/data", "", true, nil) |
| 108 | next.ImportGate = rec.Seed.VerifyHash |
| 109 | if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *stored, |
| 110 | RecordKey: RecordKey(rec.Account, rec.App, rec.Host)}, next); err != nil { |
| 111 | t.Fatal(err) |
| 112 | } |
| 113 | if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| 114 | t.Fatal(err) |
| 115 | } |
| 116 | r, _ := GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| 117 | if r.Seed.State != seeds.StateSwapped { |
| 118 | t.Fatalf("state = %s, want the CAS redone", r.Seed.State) |
| 119 | } |
| 120 | if got := mustGet(t, reg, rec.Host); got.Kind != "instance" { |
| 121 | t.Fatalf("the swapped row was refused or changed: %+v", got) |
| 122 | } |
| 123 | } |
| 124 | |
| 125 | func TestCommitClearsTheGateAndTheJournal(t *testing.T) { |
| 126 | s, sqid := shopSetup(t) |
| 127 | rec := importRecord(t, s, sqid) |
| 128 | reg := regWith(t, addrRow(rec, sqid)) |
| 129 | cfg := importCfg(t, nil) |
| 130 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 131 | t.Fatal(err) |
| 132 | } |
| 133 | for _, step := range []func(string) func(*Record) error{SeedPlace, SeedCommit} { |
| 134 | if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, step(rec.Seed.ID)); err != nil { |
| 135 | t.Fatal(err) |
| 136 | } |
| 137 | } |
| 138 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 139 | t.Fatal(err) |
| 140 | } |
| 141 | got := mustGet(t, reg, rec.Host) |
| 142 | if got.ImportGate != "" { |
| 143 | t.Fatalf("gate after commit = %q", got.ImportGate) |
| 144 | } |
| 145 | if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| 146 | t.Fatalf("journal after commit = %+v", j) |
| 147 | } |
| 148 | // The same pass converged the row: its seed column says committed. |
| 149 | if col, _ := seeds.ParseRowSeed(got.SeedSource); col == nil || col.State != seeds.StateCommitted { |
| 150 | t.Fatalf("seed column after commit = %+v", col) |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // Each record state a journal row at swapped can meet (the Task 4 table, |
| 155 | // spec §5.4 "Crashes" and §5.6). A rollback case must leave the host to |
| 156 | // the rollback alone: no ordinary handling, no removal, and none of the |
| 157 | // no-journal paths. |
| 158 | func TestJournalDecidesEachRecordState(t *testing.T) { |
| 159 | type want int |
| 160 | const ( |
| 161 | rollback want = iota |
| 162 | nothing |
| 163 | ordinary |
| 164 | ) |
| 165 | for _, c := range []struct { |
| 166 | name string |
| 167 | state string |
| 168 | edit func(t *testing.T, x *journaled) |
| 169 | want want |
| 170 | // for a rollback: the record's seed state after it, and whether |
| 171 | // the host's replica was this import's to delete |
| 172 | after string |
| 173 | replicaGone bool |
| 174 | }{ |
| 175 | {name: "record gone", state: seeds.StateReady, want: rollback, replicaGone: true, edit: func(t *testing.T, x *journaled) { |
| 176 | if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| 177 | t.Fatal(err) |
| 178 | } |
| 179 | }}, |
| 180 | {name: "record with no seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) { |
| 181 | plain := x.rec |
| 182 | plain.Seed = nil |
| 183 | putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain)) |
| 184 | }}, |
| 185 | {name: "another seed", state: seeds.StateReady, want: rollback, edit: func(t *testing.T, x *journaled) { |
| 186 | other := x.rec |
| 187 | seed := *x.rec.Seed |
| 188 | seed.ID, _ = seeds.NewID() |
| 189 | other.Seed = &seed |
| 190 | putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, other)) |
| 191 | }}, |
| 192 | {name: "aborted", state: seeds.StateAborted, want: rollback, after: seeds.StateRolledBack, replicaGone: true}, |
| 193 | // m-c: a refusal that lost the race to this box's swap. The |
| 194 | // journal wins. |
| 195 | {name: "refused", state: seeds.StateRefused, want: rollback, after: seeds.StateRefused, replicaGone: true}, |
| 196 | {name: "rolled back", state: seeds.StateRolledBack, want: rollback, after: seeds.StateRolledBack, replicaGone: true}, |
| 197 | // m12: states that come before any swap. Unreachable, so the |
| 198 | // host is held and logged. |
| 199 | {name: "reserving", state: seeds.StateReserving, want: nothing}, |
| 200 | {name: "awaiting upload", state: seeds.StateAwaitingUpload, want: nothing}, |
| 201 | {name: "uploaded", state: seeds.StateUploaded, want: nothing}, |
| 202 | {name: "swapped", state: seeds.StateSwapped, want: ordinary}, |
| 203 | {name: "placed", state: seeds.StatePlaced, want: ordinary}, |
| 204 | } { |
| 205 | t.Run(c.name, func(t *testing.T) { |
| 206 | x := journaledImport(t, c.state) |
| 207 | if c.edit != nil { |
| 208 | c.edit(t, &x) |
| 209 | } |
| 210 | b := x.box(t) |
| 211 | before, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host) |
| 212 | if c.want != rollback { |
| 213 | // Converging the row as an ordinary owned row would rewrite |
| 214 | // its seed column: make it stale so ordinary handling shows. |
| 215 | // (A rollback acts only on the import's row as the swap |
| 216 | // wrote it, so its cases keep the column.) |
| 217 | if err := x.reg.SetSeedSource(x.rec.Host, "{}"); err != nil { |
| 218 | t.Fatal(err) |
| 219 | } |
| 220 | x.row.SeedSource = "{}" |
| 221 | } |
| 222 | logged := captureLog(t) |
| 223 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 224 | t.Fatal(err) |
| 225 | } |
| 226 | after, _ := GetRecord(ctx, x.s, x.rec.Account, x.rec.App, x.rec.Host) |
| 227 | switch c.want { |
| 228 | case rollback: |
| 229 | x.rolledBack(t, b, c.replicaGone) |
| 230 | if c.after != "" { |
| 231 | if after == nil || after.Seed == nil || after.Seed.State != c.after { |
| 232 | t.Fatalf("record after the rollback = %+v, want seed %s", after, c.after) |
| 233 | } |
| 234 | } else if !reflect.DeepEqual(after, before) { |
| 235 | t.Fatalf("record changed: %+v -> %+v", before, after) |
| 236 | } |
| 237 | case nothing: |
| 238 | x.untouched(t) |
| 239 | if len(b.steps) != 0 { |
| 240 | t.Fatalf("a rollback started: %q", b.steps) |
| 241 | } |
| 242 | if !strings.Contains(logged.String(), "is journaled but its record says "+c.state) { |
| 243 | t.Fatalf("the host was not held with a log line:\n%s", logged) |
| 244 | } |
| 245 | if !reflect.DeepEqual(after, before) { |
| 246 | t.Fatalf("record changed: %+v -> %+v", before, after) |
| 247 | } |
| 248 | case ordinary: |
| 249 | if len(b.steps) != 0 { |
| 250 | t.Fatalf("rolled back a live import: %q", b.steps) |
| 251 | } |
| 252 | got := mustGet(t, x.reg, x.rec.Host) |
| 253 | if got.SeedSource != SeedColumn(x.rec) || got.ImportGate != x.rec.Seed.VerifyHash { |
| 254 | t.Fatalf("row = %+v, want it converged and still gated", got) |
| 255 | } |
| 256 | if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || !reflect.DeepEqual(*j, x.j) { |
| 257 | t.Fatalf("journal changed: %+v", j) |
| 258 | } |
| 259 | } |
| 260 | }) |
| 261 | } |
| 262 | } |
| 263 | |
| 264 | // The redo and the commit act only for the record the journal names. A |
| 265 | // record holding this seed ID under another key moves nothing forward; |
| 266 | // the journal's own record is the one that decides, and here it is gone, |
| 267 | // so the import is rolled back. The record under the other key holds the |
| 268 | // host, so the host's replica is left to it, and the record is not moved. |
| 269 | func TestJournalNeedsItsRecordKeyToMoveForward(t *testing.T) { |
| 270 | for _, state := range []string{seeds.StateReady, seeds.StateCommitted} { |
| 271 | t.Run(state, func(t *testing.T) { |
| 272 | s, sqid := shopSetup(t) |
| 273 | rec := newImport(sqid) |
| 274 | rec.Seed.State = state |
| 275 | if err := CreateRecord(ctx, s, rec); err != nil { |
| 276 | t.Fatal(err) |
| 277 | } |
| 278 | reg := regWith(t, addrRow(rec, sqid)) |
| 279 | original := mustGet(t, reg, rec.Host) |
| 280 | if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, Original: *original, |
| 281 | RecordKey: RecordKey(sqid, "other", rec.Host), At: 1}, importRow(rec, "/data", nil)); err != nil { |
| 282 | t.Fatal(err) |
| 283 | } |
| 284 | cfg := importCfg(t, nil) |
| 285 | b := newRollbackBox(t, reg, *original) |
| 286 | b.hooks(cfg.Seeds) |
| 287 | if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| 288 | t.Fatal(err) |
| 289 | } |
| 290 | if st := seedState(t, s, rec); st.State != state { |
| 291 | t.Fatalf("record = %s, want %s", st.State, state) |
| 292 | } |
| 293 | if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) { |
| 294 | t.Fatalf("row = %+v, want the original", got) |
| 295 | } |
| 296 | if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| 297 | t.Fatalf("journal = %+v", j) |
| 298 | } |
| 299 | if b.local || !b.replica { |
| 300 | t.Fatalf("local %v replica %v: steps %q", b.local, b.replica, b.steps) |
| 301 | } |
| 302 | }) |
| 303 | } |
| 304 | } |
| 305 | |
| 306 | // m3: a dry run redoes no CAS and clears no gate. |
| 307 | func TestJournalDryRunWritesNothing(t *testing.T) { |
| 308 | for _, state := range []string{seeds.StateReady, seeds.StateCommitted} { |
| 309 | t.Run(state, func(t *testing.T) { |
| 310 | x := journaledImport(t, state) |
| 311 | x.cfg.DryRun = true |
| 312 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 313 | t.Fatal(err) |
| 314 | } |
| 315 | if st := seedState(t, x.s, x.rec); st.State != state { |
| 316 | t.Fatalf("record = %s, want %s", st.State, state) |
| 317 | } |
| 318 | x.untouched(t) |
| 319 | }) |
| 320 | } |
| 321 | } |
| 322 | |
| 323 | // m4: a dry run starts no rollback either. |
| 324 | func TestJournalDryRunStartsNoRollback(t *testing.T) { |
| 325 | x := journaledImport(t, seeds.StateSwapped) |
| 326 | b := x.box(t) |
| 327 | x.cfg.DryRun = true |
| 328 | if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| 329 | t.Fatal(err) |
| 330 | } |
| 331 | logged := captureLog(t) |
| 332 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 333 | t.Fatal(err) |
| 334 | } |
| 335 | if len(b.steps) != 0 || !strings.Contains(logged.String(), "dry-run: would roll back "+x.rec.Host) { |
| 336 | t.Fatalf("a dry run started a rollback (%q), or did not say it would:\n%s", b.steps, logged) |
| 337 | } |
| 338 | x.untouched(t) |
| 339 | } |
| 340 | |
| 341 | // I1 (Task 4 review): another record for the journal's host never decides |
| 342 | // the journal's import. Here an app named other, under the same account, |
| 343 | // holds the host in the pass, while the import's own record is committed. |
| 344 | // The import is committed, never rolled back, and the other record never |
| 345 | // touches the row. |
| 346 | func TestJournalIgnoresAnotherRecordForItsHost(t *testing.T) { |
| 347 | for _, c := range []struct { |
| 348 | name string |
| 349 | state string |
| 350 | edit func(t *testing.T, x *journaled) |
| 351 | dup bool // the pass logs the duplicate host, keeping the other record |
| 352 | }{ |
| 353 | {name: "another record wins the duplicate pick", state: seeds.StateCommitted, dup: true}, |
| 354 | {name: "another record wins the duplicate pick, swapped", state: seeds.StateSwapped, dup: true}, |
| 355 | {name: "the import's record is out of scope", state: seeds.StateCommitted, edit: func(t *testing.T, x *journaled) { |
| 356 | moved := x.rec |
| 357 | moved.Region = "us" |
| 358 | putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved)) |
| 359 | }}, |
| 360 | } { |
| 361 | t.Run(c.name, func(t *testing.T) { |
| 362 | x := journaledImport(t, c.state) |
| 363 | if c.edit != nil { |
| 364 | c.edit(t, &x) |
| 365 | } |
| 366 | other := shopRecord(x.rec.Account) |
| 367 | other.App = "other" // listed before shop, so it wins a duplicate host |
| 368 | // A run on the other record makes the loop's handling of it |
| 369 | // visible (RunSeen): the loop must take up no other record for |
| 370 | // a host the journal owns. |
| 371 | other.ActiveRun = &RunState{ID: "run_abcdefghijklmnopqrstuvwxyz", State: RunQueued, Deadline: time.Now().Add(time.Minute)} |
| 372 | var seen []string |
| 373 | x.cfg.RunSeen = func(r Record) { seen = append(seen, RecordKey(r.Account, r.App, r.Host)) } |
| 374 | otherKey := RecordKey(other.Account, other.App, other.Host) |
| 375 | if _, err := x.s.Put(ctx, otherKey, mustJSON(t, other), ""); err != nil { |
| 376 | t.Fatal(err) |
| 377 | } |
| 378 | b := x.box(t) |
| 379 | logged := captureLog(t) |
| 380 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 381 | t.Fatal(err) |
| 382 | } |
| 383 | out := logged.String() |
| 384 | if c.dup && !strings.Contains(out, "duplicate host "+x.rec.Host+": keeping "+otherKey) { |
| 385 | t.Fatalf("the scene did not put the other record in the pass:\n%s", out) |
| 386 | } |
| 387 | if len(b.steps) != 0 || !b.local || !b.replica { |
| 388 | t.Fatalf("another record sent the import to rollback: %q\n%s", b.steps, out) |
| 389 | } |
| 390 | if len(seen) != 0 { |
| 391 | t.Fatalf("the loop took up %v for a host the journal owns", seen) |
| 392 | } |
| 393 | if c.state == seeds.StateSwapped { |
| 394 | x.untouched(t) |
| 395 | return |
| 396 | } |
| 397 | got := mustGet(t, x.reg, x.rec.Host) |
| 398 | if got.ImportGate != "" || got.App != "shop" || got.Account != x.rec.Account || got.Kind != "instance" || !got.Provisioned { |
| 399 | t.Fatalf("row = %+v, want the import's row, committed", got) |
| 400 | } |
| 401 | if got.SeedSource != x.row.SeedSource || got.Socket != x.row.Socket { |
| 402 | t.Fatalf("the row was converged by another record: %+v", got) |
| 403 | } |
| 404 | if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil { |
| 405 | t.Fatalf("journal = %+v, want it dropped by the commit", j) |
| 406 | } |
| 407 | }) |
| 408 | } |
| 409 | } |
| 410 | |
| 411 | // getFails is a store whose reads fail with something other than "not |
| 412 | // found". |
| 413 | type getFails struct { |
| 414 | *release.DirStore |
| 415 | } |
| 416 | |
| 417 | func (getFails) Get(context.Context, string) ([]byte, string, error) { |
| 418 | return nil, "", errors.New("connection reset") |
| 419 | } |
| 420 | |
| 421 | // m1, m2 (Task 4 review): every way journalRecord cannot tell decides |
| 422 | // nothing (known false), and only "not found" under the journal's own key |
| 423 | // means gone. |
| 424 | func TestJournalRecordFailsClosed(t *testing.T) { |
| 425 | s, sqid := shopSetup(t) |
| 426 | rec := newImport(sqid) |
| 427 | key := RecordKey(rec.Account, rec.App, rec.Host) |
| 428 | j := registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, RecordKey: key, State: registry.JournalSwapped} |
| 429 | other := shopRecord(sqid) |
| 430 | other.App = "other" |
| 431 | inPass := map[string]Record{rec.Host: rec} |
| 432 | otherInPass := map[string]Record{rec.Host: other} |
| 433 | none := map[string]bool{} |
| 434 | |
| 435 | for _, c := range []struct { |
| 436 | name string |
| 437 | store release.Store |
| 438 | j registry.ImportJournal |
| 439 | recs map[string]Record |
| 440 | readFailed bool |
| 441 | protect map[string]bool |
| 442 | body []byte // written under the journal's key first |
| 443 | known bool |
| 444 | gone bool |
| 445 | pass bool |
| 446 | }{ |
| 447 | {name: "the pass's own record", store: s, j: j, recs: inPass, protect: none, known: true, pass: true}, |
| 448 | {name: "the pass's own record, another read failed", store: s, j: j, recs: inPass, readFailed: true, protect: none, known: true, pass: true}, |
| 449 | {name: "protected", store: s, j: j, recs: inPass, protect: map[string]bool{rec.Host: true}}, |
| 450 | {name: "no record key", store: s, j: registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID, State: registry.JournalSwapped}, recs: map[string]Record{}, protect: none}, |
| 451 | {name: "missing from a pass that failed a read", store: s, j: j, recs: map[string]Record{}, readFailed: true, protect: none}, |
| 452 | {name: "another record in the pass, a read failed", store: s, j: j, recs: otherInPass, readFailed: true, protect: none}, |
| 453 | {name: "the read fails", store: getFails{s}, j: j, recs: otherInPass, protect: none}, |
| 454 | {name: "the body does not decode", store: s, j: j, recs: otherInPass, protect: none, body: []byte("{")}, |
| 455 | {name: "the body names another host", store: s, j: j, recs: otherInPass, protect: none, body: func() []byte { |
| 456 | r := rec |
| 457 | r.Host = "b.shop.example.com" |
| 458 | return mustJSON(t, r) |
| 459 | }()}, |
| 460 | {name: "gone", store: s, j: j, recs: otherInPass, protect: none, known: true, gone: true}, |
| 461 | {name: "out of the pass, read by its key", store: s, j: j, recs: otherInPass, protect: none, body: mustJSON(t, rec), known: true}, |
| 462 | } { |
| 463 | t.Run(c.name, func(t *testing.T) { |
| 464 | _ = s.Delete(ctx, key) |
| 465 | if c.body != nil { |
| 466 | if _, err := s.Put(ctx, key, c.body, ""); err != nil { |
| 467 | t.Fatal(err) |
| 468 | } |
| 469 | } |
| 470 | got, pass, known := journalRecord(ctx, c.store, c.j, c.recs, c.readFailed, c.protect) |
| 471 | if known != c.known || pass != c.pass || (known && (got == nil) != c.gone) { |
| 472 | t.Fatalf("record %v, in pass %v, known %v; want known %v, gone %v, in pass %v", got, pass, known, c.known, c.gone, c.pass) |
| 473 | } |
| 474 | if got != nil && RecordKey(got.Account, got.App, got.Host) != key { |
| 475 | t.Fatalf("answered with %s, not the journal's record", RecordKey(got.Account, got.App, got.Host)) |
| 476 | } |
| 477 | }) |
| 478 | } |
| 479 | } |
| 480 | |
| 481 | // The removal phase never removes a journaled host's row, whatever the |
| 482 | // journal's state: the journal owns it until the row is gone (M1, M13). |
| 483 | // The gated row is Provisioned and its record is gone, which without the |
| 484 | // journal is exactly what the removal phase deletes. |
| 485 | func TestRemovalSkipsJournaledHosts(t *testing.T) { |
| 486 | for _, state := range []string{registry.JournalSwapped, registry.JournalRollingBack} { |
| 487 | t.Run(state, func(t *testing.T) { |
| 488 | x := journaledImport(t, seeds.StateSwapped) |
| 489 | if state == registry.JournalRollingBack { |
| 490 | if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 7); err != nil { |
| 491 | t.Fatal(err) |
| 492 | } |
| 493 | x.row = *mustGet(t, x.reg, x.rec.Host) |
| 494 | j, _ := x.reg.ImportJournal(x.rec.Host) |
| 495 | x.j = *j |
| 496 | } |
| 497 | if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| 498 | t.Fatal(err) |
| 499 | } |
| 500 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 501 | t.Fatal(err) |
| 502 | } |
| 503 | x.untouched(t) |
| 504 | }) |
| 505 | } |
| 506 | } |
| 507 | |
| 508 | // M1: the journal is read even with seed hooks unarmed. Its hosts are |
| 509 | // protected from removal and from ordinary handling, and nothing is done |
| 510 | // for them, a commit included. |
| 511 | func TestJournalIsReadWithSeedHooksUnarmed(t *testing.T) { |
| 512 | t.Run("record gone", func(t *testing.T) { |
| 513 | x := journaledImport(t, seeds.StateSwapped) |
| 514 | x.cfg.Seeds = nil |
| 515 | if err := x.s.Delete(ctx, RecordKey(x.rec.Account, x.rec.App, x.rec.Host)); err != nil { |
| 516 | t.Fatal(err) |
| 517 | } |
| 518 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 519 | t.Fatal(err) |
| 520 | } |
| 521 | x.untouched(t) |
| 522 | }) |
| 523 | t.Run("record replaced by one with no seed", func(t *testing.T) { |
| 524 | x := journaledImport(t, seeds.StateSwapped) |
| 525 | x.cfg.Seeds = nil |
| 526 | plain := x.rec |
| 527 | plain.Seed = nil |
| 528 | putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, plain)) |
| 529 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 530 | t.Fatal(err) |
| 531 | } |
| 532 | x.untouched(t) |
| 533 | }) |
| 534 | t.Run("committed", func(t *testing.T) { |
| 535 | x := journaledImport(t, seeds.StateCommitted) |
| 536 | x.cfg.Seeds = nil |
| 537 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 538 | t.Fatal(err) |
| 539 | } |
| 540 | // Neither committed nor rolled back: the gated row and the |
| 541 | // journal row exactly as they were. |
| 542 | x.untouched(t) |
| 543 | }) |
| 544 | } |
| 545 | |
| 546 | // corruptJournal makes ImportJournals fail: the row's original no longer |
| 547 | // parses. |
| 548 | func corruptJournal(t *testing.T, path, host string) { |
| 549 | t.Helper() |
| 550 | db, err := sql.Open("sqlite", path) |
| 551 | if err != nil { |
| 552 | t.Fatal(err) |
| 553 | } |
| 554 | defer db.Close() |
| 555 | if _, err := db.Exec(`UPDATE import_journal SET original = '{' WHERE host = ?`, host); err != nil { |
| 556 | t.Fatal(err) |
| 557 | } |
| 558 | } |
| 559 | |
| 560 | // regAt is regWith at a path the test can open beside the registry. |
| 561 | func regAt(t *testing.T, routes ...registry.Route) (*registry.Registry, string) { |
| 562 | t.Helper() |
| 563 | path := filepath.Join(t.TempDir(), "registry.db") |
| 564 | reg, err := registry.Open(path) |
| 565 | if err != nil { |
| 566 | t.Fatal(err) |
| 567 | } |
| 568 | for _, rt := range routes { |
| 569 | if err := reg.Upsert(rt); err != nil { |
| 570 | t.Fatal(err) |
| 571 | } |
| 572 | } |
| 573 | return reg, path |
| 574 | } |
| 575 | |
| 576 | // M1: a journal that cannot be read is a failed read. Nothing is removed |
| 577 | // that pass, and no import is acted on: no swap, and no no-journal finish |
| 578 | // of an aborted import (it might have a journal row). |
| 579 | func TestJournalReadFailureRemovesNothingAndActsOnNoImport(t *testing.T) { |
| 580 | s, sqid := shopSetup(t) |
| 581 | // A swapped import, whose journal row will not parse. |
| 582 | swapped := importRecord(t, s, sqid) |
| 583 | // A ready import on another host, with its addr row. |
| 584 | ready := newImport(sqid) |
| 585 | ready.Host, ready.Slug = "b.shop.example.com", "b" |
| 586 | if err := CreateRecord(ctx, s, ready); err != nil { |
| 587 | t.Fatal(err) |
| 588 | } |
| 589 | // An aborted import on a third host. |
| 590 | aborted := newImport(sqid) |
| 591 | aborted.Host, aborted.Slug = "c.shop.example.com", "c" |
| 592 | aborted.Seed.State = seeds.StateAborted |
| 593 | if err := CreateRecord(ctx, s, aborted); err != nil { |
| 594 | t.Fatal(err) |
| 595 | } |
| 596 | // And a provisioned row whose record is gone. |
| 597 | orphan := RouteFor(shopRecord(sqid), "/data", "", true, nil) |
| 598 | orphan.Host, orphan.Slug = "d.shop.example.com", "d" |
| 599 | reg, path := regAt(t, addrRow(swapped, sqid), addrRow(ready, sqid), orphan) |
| 600 | original := mustGet(t, reg, swapped.Host) |
| 601 | if err := reg.SwapForImport(registry.ImportJournal{Host: swapped.Host, SeedID: swapped.Seed.ID, Original: *original, |
| 602 | RecordKey: RecordKey(swapped.Account, swapped.App, swapped.Host), At: 1}, importRow(swapped, "/data", nil)); err != nil { |
| 603 | t.Fatal(err) |
| 604 | } |
| 605 | if err := s.Delete(ctx, RecordKey(swapped.Account, swapped.App, swapped.Host)); err != nil { |
| 606 | t.Fatal(err) |
| 607 | } |
| 608 | gated := mustGet(t, reg, swapped.Host) |
| 609 | readyRow := mustGet(t, reg, ready.Host) |
| 610 | corruptJournal(t, path, swapped.Host) |
| 611 | |
| 612 | if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| 613 | t.Fatal(err) |
| 614 | } |
| 615 | if got := mustGet(t, reg, swapped.Host); !reflect.DeepEqual(*got, *gated) { |
| 616 | t.Fatalf("the journaled row changed: %+v", got) |
| 617 | } |
| 618 | mustGet(t, reg, orphan.Host) // not removed: the pass could not read the journal |
| 619 | if got := mustGet(t, reg, ready.Host); !reflect.DeepEqual(*got, *readyRow) { |
| 620 | t.Fatalf("the ready import was swapped: %+v", got) |
| 621 | } |
| 622 | if st := seedState(t, s, ready); st.State != seeds.StateReady || st.Reason != "" { |
| 623 | t.Fatalf("ready import = %s %q", st.State, st.Reason) |
| 624 | } |
| 625 | if st := seedState(t, s, aborted); st.State != seeds.StateAborted { |
| 626 | t.Fatalf("aborted import = %s, want it left aborted", st.State) |
| 627 | } |
| 628 | } |
| 629 | |
| 630 | // A journal row's record that this box's scope excludes is still there: |
| 631 | // it is read by its key, never taken as gone, and a committed import is |
| 632 | // committed, never rolled back. |
| 633 | func TestJournalReadsAnOutOfScopeRecordByItsKey(t *testing.T) { |
| 634 | x := journaledImport(t, seeds.StateCommitted) |
| 635 | moved := x.rec |
| 636 | moved.Region = "us" |
| 637 | putReplace(t, x.s, RecordKey(x.rec.Account, x.rec.App, x.rec.Host), mustJSON(t, moved)) |
| 638 | b := x.box(t) |
| 639 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 640 | t.Fatal(err) |
| 641 | } |
| 642 | if len(b.steps) != 0 || !b.local || !b.replica { |
| 643 | t.Fatalf("rolled back a committed import: %q", b.steps) |
| 644 | } |
| 645 | if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate != "" { |
| 646 | t.Fatalf("gate = %q, want it cleared", got.ImportGate) |
| 647 | } |
| 648 | if j, _ := x.reg.ImportJournal(x.rec.Host); j != nil { |
| 649 | t.Fatalf("journal = %+v", j) |
| 650 | } |
| 651 | } |
| 652 | |
| 653 | // m-d: before commit, only the swap's gated row is the import's row. An |
| 654 | // ungated row for a swapped or placed import, hand-added or with its gate |
| 655 | // cleared by hand, is neither adopted nor converged. |
| 656 | func TestReconcileLeavesAnUngatedRowOfAnUncommittedImport(t *testing.T) { |
| 657 | for _, state := range []string{seeds.StateSwapped, seeds.StatePlaced} { |
| 658 | for _, provisioned := range []bool{false, true} { |
| 659 | name := state + "/hand-added" |
| 660 | if provisioned { |
| 661 | name = state + "/gate cleared" |
| 662 | } |
| 663 | t.Run(name, func(t *testing.T) { |
| 664 | s, sqid := shopSetup(t) |
| 665 | rec := newImport(sqid) |
| 666 | rec.Seed.State = state |
| 667 | if err := CreateRecord(ctx, s, rec); err != nil { |
| 668 | t.Fatal(err) |
| 669 | } |
| 670 | row := importRow(rec, "/data", nil) |
| 671 | row.ImportGate = "" |
| 672 | row.Provisioned = provisioned |
| 673 | row.SeedSource = "{}" // stale: a converge would rewrite it |
| 674 | reg := regWith(t, row) |
| 675 | stored := mustGet(t, reg, rec.Host) |
| 676 | if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| 677 | t.Fatal(err) |
| 678 | } |
| 679 | if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *stored) { |
| 680 | t.Fatalf("row changed:\n got %+v\nwant %+v", *got, *stored) |
| 681 | } |
| 682 | }) |
| 683 | } |
| 684 | } |
| 685 | } |
| 686 | |
| 687 | // The journal check inside swapImport stays as a backstop (Task 3's |
| 688 | // departure 1): called with this seed's journal row at swapped, it only |
| 689 | // redoes the record CAS, even with no journal pass ahead of it. |
| 690 | func TestSwapImportBackstopRedoesOnlyTheCAS(t *testing.T) { |
| 691 | x := journaledImport(t, seeds.StateReady) |
| 692 | swapImport(ctx, x.s, x.reg, x.cfg, x.rec, &x.row, func() (bool, error) { return true, nil }, nil) |
| 693 | if st := seedState(t, x.s, x.rec); st.State != seeds.StateSwapped || st.Reason != "" { |
| 694 | t.Fatalf("record = %s %q, want swapped", st.State, st.Reason) |
| 695 | } |
| 696 | x.untouched(t) |
| 697 | } |
| 698 | |
| 699 | // convergeScene is a --from seed whose record is placed and whose row on |
| 700 | // this box still holds the manifest's seed command and no seed column: |
| 701 | // the converge must set the column and clear the command. |
| 702 | func convergeScene(t *testing.T) (*release.DirStore, *registry.Registry, string, Record) { |
| 703 | t.Helper() |
| 704 | s := dirStore(t) |
| 705 | sqid := writeAccountManifest(t, s, acctID, "Shop") |
| 706 | if err := PutManifest(ctx, s, sqid, "shop", Manifest{Schema: 1, App: "shop", Domain: "shop.example.com", Seed: []string{"seed"}}); err != nil { |
| 707 | t.Fatal(err) |
| 708 | } |
| 709 | rec := shopRecord(sqid) |
| 710 | id, _ := seeds.NewID() |
| 711 | rec.Seed = &SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StatePlaced, SHA256: strings.Repeat("a", 64), Bytes: 10} |
| 712 | if err := CreateRecord(ctx, s, rec); err != nil { |
| 713 | t.Fatal(err) |
| 714 | } |
| 715 | row := RouteFor(rec, "/data", `["seed"]`, true, nil) |
| 716 | row.SeedSource = "" |
| 717 | reg, path := regAt(t, row) |
| 718 | return s, reg, path, rec |
| 719 | } |
| 720 | |
| 721 | func addTrigger(t *testing.T, path, ddl string) { |
| 722 | t.Helper() |
| 723 | db, err := sql.Open("sqlite", path) |
| 724 | if err != nil { |
| 725 | t.Fatal(err) |
| 726 | } |
| 727 | defer db.Close() |
| 728 | if _, err := db.Exec(ddl); err != nil { |
| 729 | t.Fatal(err) |
| 730 | } |
| 731 | } |
| 732 | |
| 733 | // M10: the converge sets the seed column before it clears the seed |
| 734 | // command, so the row is never left holding neither (which could boot |
| 735 | // empty). The trigger refuses that state; in the other order the seed |
| 736 | // write is refused and the command stays. |
| 737 | func TestConvergeSetsTheSeedSourceBeforeClearingTheSeed(t *testing.T) { |
| 738 | s, reg, path, rec := convergeScene(t) |
| 739 | addTrigger(t, path, `CREATE TRIGGER neither BEFORE UPDATE ON routes |
| 740 | WHEN NEW.seed = '' AND NEW.seed_source = '' |
| 741 | BEGIN SELECT RAISE(ABORT, 'a row with neither a seed command nor a seed source'); END`) |
| 742 | if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| 743 | t.Fatal(err) |
| 744 | } |
| 745 | if got := mustGet(t, reg, rec.Host); got.Seed != "" || got.SeedSource != SeedColumn(rec) { |
| 746 | t.Fatalf("row = seed %q source %q, want no seed command and this seed's column", got.Seed, got.SeedSource) |
| 747 | } |
| 748 | } |
| 749 | |
| 750 | // M10: a seed column that cannot be written leaves the seed command in |
| 751 | // place, so the row keeps the state it had. |
| 752 | func TestConvergeKeepsTheSeedWhenTheSeedSourceFails(t *testing.T) { |
| 753 | s, reg, path, rec := convergeScene(t) |
| 754 | addTrigger(t, path, `CREATE TRIGGER nosource BEFORE UPDATE OF seed_source ON routes |
| 755 | BEGIN SELECT RAISE(ABORT, 'seed_source is read-only here'); END`) |
| 756 | if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil { |
| 757 | t.Fatal(err) |
| 758 | } |
| 759 | if got := mustGet(t, reg, rec.Host); got.Seed != `["seed"]` || got.SeedSource != "" { |
| 760 | t.Fatalf("row = seed %q source %q, want it as it was", got.Seed, got.SeedSource) |
| 761 | } |
| 762 | } |
| 763 | |
| 764 | // Merge review S2 (b): on a box with claim control on, the journal pass |
| 765 | // does nothing but protect. A journal row there is an import a claim |
| 766 | // caught mid-flight; the box arms no swap or rollback and the claim |
| 767 | // console refuses commit, abort and finish, so no single step is safe to |
| 768 | // take. In every state the row, the journal row and the record are exactly |
| 769 | // as they were: the swap's record CAS is not redone, a committed import's |
| 770 | // gate is not cleared, and no rollback step runs, even with the rollback's |
| 771 | // hooks wired. The removal phase leaves the row too. |
| 772 | func TestJournalPassOnlyProtectsWithClaimControlOn(t *testing.T) { |
| 773 | for _, c := range []struct { |
| 774 | name string |
| 775 | record string // the record's seed state; "" deletes the record |
| 776 | journal string // the journal row's state |
| 777 | }{ |
| 778 | {"ready: the record CAS would be redone", seeds.StateReady, registry.JournalSwapped}, |
| 779 | {"swapped", seeds.StateSwapped, registry.JournalSwapped}, |
| 780 | {"placed", seeds.StatePlaced, registry.JournalSwapped}, |
| 781 | {"committed: the gate would be cleared", seeds.StateCommitted, registry.JournalSwapped}, |
| 782 | {"aborted: a rollback would start", seeds.StateAborted, registry.JournalSwapped}, |
| 783 | {"record gone: a rollback would start", "", registry.JournalSwapped}, |
| 784 | {"rolling-back", seeds.StateAborted, registry.JournalRollingBack}, |
| 785 | {"rolling-back, record gone", "", registry.JournalRollingBack}, |
| 786 | {"cleanup-pending", seeds.StateAborted, registry.JournalCleanupPending}, |
| 787 | {"cleanup-pending, record gone", "", registry.JournalCleanupPending}, |
| 788 | } { |
| 789 | t.Run(c.name, func(t *testing.T) { |
| 790 | state := c.record |
| 791 | if state == "" { |
| 792 | state = seeds.StateSwapped |
| 793 | } |
| 794 | x := journaledImport(t, state) |
| 795 | if c.journal != registry.JournalSwapped { |
| 796 | if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 7); err != nil { |
| 797 | t.Fatal(err) |
| 798 | } |
| 799 | } |
| 800 | if c.journal == registry.JournalCleanupPending { |
| 801 | if err := x.reg.RestoreImportRoute(x.rec.Host, x.rec.Seed.ID); err != nil { |
| 802 | t.Fatal(err) |
| 803 | } |
| 804 | } |
| 805 | x.row = *mustGet(t, x.reg, x.rec.Host) |
| 806 | j, _ := x.reg.ImportJournal(x.rec.Host) |
| 807 | x.j = *j |
| 808 | key := RecordKey(x.rec.Account, x.rec.App, x.rec.Host) |
| 809 | if c.record == "" { |
| 810 | if err := x.s.Delete(ctx, key); err != nil { |
| 811 | t.Fatal(err) |
| 812 | } |
| 813 | } |
| 814 | before, _, _ := x.s.Get(ctx, key) |
| 815 | b := x.box(t) |
| 816 | x.cfg.ClaimEnabled = true |
| 817 | logged := captureLog(t) |
| 818 | for pass := 0; pass < 2; pass++ { |
| 819 | if err := Reconcile(ctx, x.s, x.reg, x.cfg); err != nil { |
| 820 | t.Fatalf("pass %d: %v", pass, err) |
| 821 | } |
| 822 | x.untouched(t) |
| 823 | if after, _, _ := x.s.Get(ctx, key); !bytes.Equal(after, before) { |
| 824 | t.Fatalf("pass %d: the record changed:\n got %s\nwant %s", pass, after, before) |
| 825 | } |
| 826 | if len(b.steps) != 0 || len(b.retired) != 0 || !b.local || !b.replica || !b.key { |
| 827 | t.Fatalf("pass %d: the box was touched: steps %q, retired %q, local %v, replica %v, key %v", |
| 828 | pass, b.steps, b.retired, b.local, b.replica, b.key) |
| 829 | } |
| 830 | } |
| 831 | if !strings.Contains(logged.String(), "claim control is on for this box; its row is protected and nothing else is done") { |
| 832 | t.Fatalf("the protection was not logged:\n%s", logged) |
| 833 | } |
| 834 | }) |
| 835 | } |
| 836 | } |
| 837 | |
| 838 | // The same journal rows with claim control off move forward as before, so |
| 839 | // the test above fails for the reason it says when the guard is removed. |
| 840 | func TestJournalPassMovesForwardWithClaimControlOff(t *testing.T) { |
| 841 | ready := journaledImport(t, seeds.StateReady) |
| 842 | if err := Reconcile(ctx, ready.s, ready.reg, ready.cfg); err != nil { |
| 843 | t.Fatal(err) |
| 844 | } |
| 845 | if st := seedState(t, ready.s, ready.rec); st.State != seeds.StateSwapped { |
| 846 | t.Fatalf("record = %s, want the swap's CAS redone", st.State) |
| 847 | } |
| 848 | committed := journaledImport(t, seeds.StateCommitted) |
| 849 | if err := Reconcile(ctx, committed.s, committed.reg, committed.cfg); err != nil { |
| 850 | t.Fatal(err) |
| 851 | } |
| 852 | if got := mustGet(t, committed.reg, committed.rec.Host); got.ImportGate != "" { |
| 853 | t.Fatalf("gate = %q, want it cleared", got.ImportGate) |
| 854 | } |
| 855 | } |
| 856 | |
amadan