amadan

carlos / platform Public

Clone 
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1package registry
2
3import (
4 "context"
5 "errors"
6 "path/filepath"
7 "reflect"
8 "testing"
9 "time"
10)
11
12// Neutral fixtures: app shop, hosts under shop.example.com. "bab" is the
13// pinned ops account (Sqid(1)); "bbw" is Sqid(42), an ordinary account.
14const (
15 importHost = "old.shop.example.com"
16 importSeedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
17 importRecordKey = "apps/bbw/shop/instances/old.shop.example.com.json"
18)
19
20func legacyRow() Route {
21 return Route{Host: importHost, Kind: "service", Addr: "127.0.0.1:8401", Status: "ready",
22 App: "shop", Account: "bab", Channel: "stable", Unit: "shop@old.service", Caps: "", CreatedAt: 1700000000}
23}
24
25func execRow() Route {
26 return Route{Host: importHost, Kind: "instance", Socket: "/run/carlos/instances/old/app.sock",
27 DBPath: "/data/instances/old/old.shop.example.com.db", Status: "ready", App: "shop", Account: "bbw",
28 Channel: "stable", Hibernate: true, Provisioned: true, ImportGate: "abc123", CreatedAt: 1700000001}
29}
30
31// swapRow is execRow as the swap writes it: its seed column names the
32// journal's seed (IsImportRow).
33func swapRow() Route {
34 r := execRow()
35 r.SeedSource = `{"id":"` + importSeedID + `","kind":"import","state":"swapped"}`
36 return r
37}
38
39func mustJournal(t *testing.T, reg *Registry, host string) *ImportJournal {
40 t.Helper()
41 j, err := reg.ImportJournal(host)
42 if err != nil {
43 t.Fatal(err)
44 }
45 return j
46}
47
48func mustDigest(t *testing.T, f func() (string, error)) string {
49 t.Helper()
50 d, err := f()
51 if err != nil {
52 t.Fatal(err)
53 }
54 return d
55}
56
57func TestSwapCommitAndRestore(t *testing.T) {
58 reg := openTestRegistry(t)
59 old := legacyRow()
60 if err := reg.Upsert(old); err != nil {
61 t.Fatal(err)
62 }
63 stored := mustRoute(t, reg, old.Host)
64 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
65 if err := reg.SwapForImport(j, swapRow()); err != nil {
66 t.Fatal(err)
67 }
68 got := mustRoute(t, reg, old.Host)
69 if got.Kind != "instance" || got.ImportGate != "abc123" || got.Account != "bbw" {
70 t.Fatalf("after swap: %+v", got)
71 }
72 jr := mustJournal(t, reg, old.Host)
73 if jr == nil || jr.State != JournalSwapped || !reflect.DeepEqual(jr.Original, stored) || jr.RecordKey != j.RecordKey || jr.At != 1 {
74 t.Fatalf("journal = %+v", jr)
75 }
76 // A second swap for the host is refused: the journal is there.
77 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
78 t.Fatalf("second swap: %v", err)
79 }
80 // Close the gate, restore the route exactly, and the journal says where it is.
81 if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
82 t.Fatal(err)
83 }
84 if got := mustRoute(t, reg, old.Host); got.ImportGate != ImportGateClosed {
85 t.Fatalf("gate = %q, want closed", got.ImportGate)
86 }
87 // Commit is only from swapped: a rolling-back import cannot commit.
88 if err := reg.CommitImport(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
89 t.Fatalf("commit while rolling back: %v", err)
90 }
91 // Closing again resumes (idempotent) and keeps the stored generation:
92 // step 2 recognises its own fence by it.
93 if err := reg.CloseImportGate(old.Host, j.SeedID, 78); err != nil {
94 t.Fatal(err)
95 }
96 if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalRollingBack || jr.FenceGen != 77 {
97 t.Fatalf("journal after a second close = %+v", jr)
98 }
99 if got := mustRoute(t, reg, old.Host); got.ImportGate != ImportGateClosed {
100 t.Fatalf("gate after a second close = %q", got.ImportGate)
101 }
102 if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {
103 t.Fatal(err)
104 }
105 if got := mustRoute(t, reg, old.Host); !reflect.DeepEqual(got, stored) {
106 t.Fatalf("restored %+v, want %+v", got, stored)
107 }
108 if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 77 {
109 t.Fatalf("journal after restore = %+v", jr)
110 }
111 // cleanup-pending never goes back: no second restore, no gate close.
112 if err := reg.RestoreImportRoute(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
113 t.Fatalf("restore from cleanup-pending: %v", err)
114 }
115 if err := reg.CloseImportGate(old.Host, j.SeedID, 78); !errors.Is(err, ErrImportJournal) {
116 t.Fatalf("close from cleanup-pending: %v", err)
117 }
118 if err := reg.DeleteImportJournal(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {
119 t.Fatalf("delete with another id: %v", err)
120 }
121 if err := reg.DeleteImportJournal(old.Host, j.SeedID); err != nil {
122 t.Fatal(err)
123 }
124 if jr := mustJournal(t, reg, old.Host); jr != nil {
125 t.Fatalf("journal after delete = %+v", jr)
126 }
127 if err := reg.DeleteImportJournal(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
128 t.Fatalf("delete with no row: %v", err)
129 }
130}
131
132func TestCommitImportClearsTheGateAndTheJournal(t *testing.T) {
133 reg := openTestRegistry(t)
134 old := legacyRow()
135 if err := reg.Upsert(old); err != nil {
136 t.Fatal(err)
137 }
138 stored := mustRoute(t, reg, old.Host)
139 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
140 if err := reg.SwapForImport(j, swapRow()); err != nil {
141 t.Fatal(err)
142 }
143 if err := reg.CommitImport(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {
144 t.Fatalf("commit with another id: %v", err)
145 }
146 before := mustDigest(t, reg.Fingerprint)
147 if err := reg.CommitImport(old.Host, j.SeedID); err != nil {
148 t.Fatal(err)
149 }
150 got := mustRoute(t, reg, old.Host)
151 jr := mustJournal(t, reg, old.Host)
152 after := mustDigest(t, reg.Fingerprint)
153 if got.ImportGate != "" || jr != nil || before == after {
154 t.Fatalf("after commit: gate %q journal %+v fingerprint moved %v", got.ImportGate, jr, before != after)
155 }
156 // The committed row is the exec row, ungated.
157 want := swapRow()
158 want.ImportGate = ""
159 if !reflect.DeepEqual(got, want) {
160 t.Fatalf("committed row %+v, want %+v", got, want)
161 }
162}
163
164func TestSwapForImportRefusesAChangedRow(t *testing.T) {
165 reg := openTestRegistry(t)
166 old := legacyRow()
167 if err := reg.Upsert(old); err != nil {
168 t.Fatal(err)
169 }
170 stored := mustRoute(t, reg, old.Host)
171 changed := stored
172 changed.Addr = "127.0.0.1:9999"
173 if err := reg.Upsert(changed); err != nil {
174 t.Fatal(err)
175 }
176 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
177 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
178 t.Fatalf("swap over a changed row: %v", err)
179 }
180 if got := mustRoute(t, reg, old.Host); got.Addr != "127.0.0.1:9999" {
181 t.Fatal("the changed row was replaced")
182 }
183 if jr := mustJournal(t, reg, old.Host); jr != nil {
184 t.Fatalf("a refused swap left a journal row: %+v", jr)
185 }
186}
187
188func TestSwapForImportRefusesAMissingRowAndAMalformedSwap(t *testing.T) {
189 reg := openTestRegistry(t)
190 old := legacyRow()
191 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: old}
192 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
193 t.Fatalf("swap with no stored row: %v", err)
194 }
195 if err := reg.Upsert(old); err != nil {
196 t.Fatal(err)
197 }
198 stored := mustRoute(t, reg, old.Host)
199 j.Original = stored
200 ungated := swapRow()
201 ungated.ImportGate = ""
202 other := swapRow()
203 other.Host = "new.shop.example.com"
204 noID := j
205 noID.SeedID = ""
206 noKey := j
207 noKey.RecordKey = ""
208 otherSeed := swapRow()
209 otherSeed.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb","kind":"import","state":"swapped"}`
210 unprovisioned := swapRow()
211 unprovisioned.Provisioned = false
212 for name, c := range map[string]struct {
213 j ImportJournal
214 next Route
215 }{
216 "an ungated row": {j, ungated},
217 "another host": {j, other},
218 "no seed id": {noID, swapRow()},
219 "no record key (N1)": {noKey, swapRow()},
220 "a row naming another seed": {j, otherSeed},
221 "a row with no seed column": {j, execRow()},
222 "a row no record provisioned": {j, unprovisioned},
223 } {
224 if err := reg.SwapForImport(c.j, c.next); !errors.Is(err, ErrImportJournal) {
225 t.Errorf("%s: %v", name, err)
226 }
227 }
228 if got := mustRoute(t, reg, old.Host); !reflect.DeepEqual(got, stored) {
229 t.Fatalf("a refused swap changed the row: %+v", got)
230 }
231 if js, err := reg.ImportJournals(); err != nil || len(js) != 0 {
232 t.Fatalf("journals = %+v, %v", js, err)
233 }
234}
235
236func TestUpsertNeverChangesTheImportGate(t *testing.T) {
237 reg := openTestRegistry(t)
238 row := execRow()
239 if err := reg.Upsert(row); err != nil {
240 t.Fatal(err)
241 }
242 row.ImportGate = ""
243 if err := reg.Upsert(row); err != nil {
244 t.Fatal(err)
245 }
246 if got := mustRoute(t, reg, row.Host); got.ImportGate != "abc123" {
247 t.Fatalf("an Upsert cleared the gate: %q", got.ImportGate)
248 }
249 row.ImportGate = "def456"
250 if err := reg.Upsert(row); err != nil {
251 t.Fatal(err)
252 }
253 if got := mustRoute(t, reg, row.Host); got.ImportGate != "abc123" {
254 t.Fatalf("an Upsert changed the gate: %q", got.ImportGate)
255 }
256}
257
258func TestImportJournalsListsEveryRow(t *testing.T) {
259 reg := openTestRegistry(t)
260 for i, host := range []string{"b.shop.example.com", "a.shop.example.com"} {
261 old := legacyRow()
262 old.Host = host
263 if err := reg.Upsert(old); err != nil {
264 t.Fatal(err)
265 }
266 stored := mustRoute(t, reg, host)
267 next := swapRow()
268 next.Host = host
269 j := ImportJournal{Host: host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: int64(i + 1)}
270 if err := reg.SwapForImport(j, next); err != nil {
271 t.Fatal(err)
272 }
273 }
274 js, err := reg.ImportJournals()
275 if err != nil {
276 t.Fatal(err)
277 }
278 if len(js) != 2 || js[0].Host != "a.shop.example.com" || js[1].Host != "b.shop.example.com" ||
279 js[0].Original.Host != "a.shop.example.com" || js[0].At != 2 || js[1].State != JournalSwapped {
280 t.Fatalf("journals = %+v", js)
281 }
282}
283
284// The gate moves both digests, and an ungated registry digests as it did
285// before the column existed (the goldens in hold_test.go cover that half).
286func TestImportGateMovesBothFingerprints(t *testing.T) {
287 reg := openTestRegistry(t)
288 if err := reg.Upsert(legacyRow()); err != nil {
289 t.Fatal(err)
290 }
291 fp1 := mustDigest(t, reg.Fingerprint)
292 sfp1 := mustDigest(t, reg.SnapshotFingerprint)
293 if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = 'abc123' WHERE host = ?`, importHost); err != nil {
294 t.Fatal(err)
295 }
296 fp2 := mustDigest(t, reg.Fingerprint)
297 sfp2 := mustDigest(t, reg.SnapshotFingerprint)
298 if fp1 == fp2 || sfp1 == sfp2 {
299 t.Fatal("setting the gate did not move both fingerprints")
300 }
301 if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, importHost); err != nil {
302 t.Fatal(err)
303 }
304 if fp3 := mustDigest(t, reg.Fingerprint); fp3 == fp2 {
305 t.Fatal("closing the gate did not move the edge's fingerprint")
306 }
307 if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, importHost); err != nil {
308 t.Fatal(err)
309 }
310 fp4 := mustDigest(t, reg.Fingerprint)
311 sfp4 := mustDigest(t, reg.SnapshotFingerprint)
312 if fp4 != fp1 || sfp4 != sfp1 {
313 t.Fatal("clearing the gate did not restore the ungated digests")
314 }
315}
316
317// A registry from before the column gains it on open, every time.
318func TestImportGateMigrationOnALegacyDB(t *testing.T) {
319 path := filepath.Join(t.TempDir(), "registry.db")
320 r, err := Open(path)
321 if err != nil {
322 t.Fatal(err)
323 }
324 if err := r.Upsert(legacyRow()); err != nil {
325 t.Fatal(err)
326 }
327 if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN import_gate`); err != nil {
328 t.Fatal(err)
329 }
330 if _, err := r.sql.Exec(`DROP TABLE import_journal`); err != nil {
331 t.Fatal(err)
332 }
333 r.Close()
334 for i := 0; i < 2; i++ {
335 r, err = Open(path)
336 if err != nil {
337 t.Fatalf("open %d: %v", i, err)
338 }
339 rt, err := r.Get(importHost)
340 if err != nil || rt == nil || rt.ImportGate != "" {
341 t.Fatalf("open %d: %+v %v", i, rt, err)
342 }
343 if js, err := r.ImportJournals(); err != nil || len(js) != 0 {
344 t.Fatalf("open %d: journals %+v %v", i, js, err)
345 }
346 r.Close()
347 }
348}
349
350// Preflight M6: the cold handover's own INSERT carries import_gate, so a
351// restored row keeps its gate.
352func TestRestoreClaimRowsCarriesTheImportGate(t *testing.T) {
353 ctx := context.Background()
354 reg := openTestRegistry(t)
355 owner := ClaimOwnership{"inc", 7, "installation-7", "route-7"}
356 row := execRow()
357 row.ClaimOwnership = owner
358 if err := reg.Upsert(row); err != nil {
359 t.Fatal(err)
360 }
361 old, err := reg.Get(row.Host)
362 if err != nil || old == nil || old.ImportGate != "abc123" {
363 t.Fatalf("stored = %+v, %v", old, err)
364 }
365 targets := []ClaimRow{{Kind: "route", Host: row.Host, Ownership: owner}}
366 if err := reg.RetireClaimRows(ctx, "handover-old", targets, func([]ClaimRowArchive) error { return nil }); err != nil {
367 t.Fatal(err)
368 }
369 next := *old
370 next.ClaimOwnership = ClaimOwnership{"inc", 8, "installation-8", "route-8"}
371 if err := reg.RestoreClaimRows(ctx, "handover", "handover-old", ColdRows{Route: *old}, ColdRows{Route: next}); err != nil {
372 t.Fatal(err)
373 }
374 got, err := reg.Get(row.Host)
375 if err != nil || got == nil {
376 t.Fatalf("restored = %+v, %v", got, err)
377 }
378 if *got != next {
379 t.Fatalf("restored %+v, want %+v", *got, next)
380 }
381 // A replay of the same handover is a success, and the gate is still there.
382 if err := reg.RestoreClaimRows(ctx, "handover", "handover-old", ColdRows{Route: *old}, ColdRows{Route: next}); err != nil {
383 t.Fatal(err)
384 }
385 if got := mustRoute(t, reg, row.Host); got != next || got.ImportGate != "abc123" {
386 t.Fatalf("after replay %+v, want %+v", got, next)
387 }
388}
389
390// Fix round 1: an empty journal digests byte-for-byte as before the journal
391// was hashed. The golden was produced by this fixture on f20d2c5e, whose
392// SnapshotFingerprint did not read import_journal.
393func TestSnapshotFingerprintUnchangedForAnEmptyJournal(t *testing.T) {
394 reg := openTestRegistry(t)
395 if err := reg.Upsert(legacyRow()); err != nil {
396 t.Fatal(err)
397 }
398 gated := execRow()
399 gated.Host = "gated.shop.example.com"
400 if err := reg.Upsert(gated); err != nil {
401 t.Fatal(err)
402 }
403 const want = "2:28920899b7d5e223b887dadf6924ee774b34d8d412f1c58bebdb214c35d82b07"
404 got, err := reg.SnapshotFingerprint()
405 if err != nil {
406 t.Fatal(err)
407 }
408 if got != want {
409 t.Fatalf("SnapshotFingerprint = %s, want %s", got, want)
410 }
411}
412
413// Fix round 1 (review Important 1): the journal is in the snapshot digest,
414// so a deleted cleanup-pending row cannot linger in the last snapshot and be
415// restored cold.
416func TestSnapshotFingerprintSeesTheJournal(t *testing.T) {
417 reg := openTestRegistry(t)
418 old := legacyRow()
419 if err := reg.Upsert(old); err != nil {
420 t.Fatal(err)
421 }
422 stored := mustRoute(t, reg, old.Host)
423 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
424 if err := reg.SwapForImport(j, swapRow()); err != nil {
425 t.Fatal(err)
426 }
427 if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
428 t.Fatal(err)
429 }
430 if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {
431 t.Fatal(err)
432 }
433 pending := mustDigest(t, reg.SnapshotFingerprint)
434 // A journal-only change moves it too.
435 if _, err := reg.sql.Exec(`UPDATE import_journal SET fence_gen = 78 WHERE host = ?`, old.Host); err != nil {
436 t.Fatal(err)
437 }
438 if mustDigest(t, reg.SnapshotFingerprint) == pending {
439 t.Fatal("a journal-only change did not move SnapshotFingerprint")
440 }
441 if _, err := reg.sql.Exec(`UPDATE import_journal SET fence_gen = 77 WHERE host = ?`, old.Host); err != nil {
442 t.Fatal(err)
443 }
444 if mustDigest(t, reg.SnapshotFingerprint) != pending {
445 t.Fatal("SnapshotFingerprint is not a function of the journal's contents")
446 }
447 routesBefore := mustDigest(t, reg.Fingerprint)
448 if err := reg.DeleteImportJournal(old.Host, j.SeedID); err != nil {
449 t.Fatal(err)
450 }
451 if mustDigest(t, reg.Fingerprint) != routesBefore {
452 t.Fatal("the journal moved the edge's fingerprint")
453 }
454 if mustDigest(t, reg.SnapshotFingerprint) == pending {
455 t.Fatal("deleting the journal row did not move SnapshotFingerprint")
456 }
457}
458
459// Review Minor 1: the swap refuses an original it could not put back as an
460// ordinary legacy row.
461func TestSwapForImportRefusesAnOwnedHeldOrGatedOriginal(t *testing.T) {
462 for name, mutate := range map[string]func(*Route){
463 "claim-owned": func(r *Route) { r.ClaimOwnership = ClaimOwnership{"inc", 7, "installation-7", "route-7"} },
464 "held": func(r *Route) { r.HoldOp = "op-1" },
465 "gated": func(r *Route) { r.ImportGate = "abc123" },
466 } {
467 t.Run(name, func(t *testing.T) {
468 reg := openTestRegistry(t)
469 old := legacyRow()
470 mutate(&old)
471 if err := reg.Upsert(old); err != nil {
472 t.Fatal(err)
473 }
474 stored := mustRoute(t, reg, old.Host)
475 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
476 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
477 t.Fatalf("swap: %v", err)
478 }
479 if got := mustRoute(t, reg, old.Host); got != stored {
480 t.Fatalf("a refused swap changed the row: %+v", got)
481 }
482 if jr := mustJournal(t, reg, old.Host); jr != nil {
483 t.Fatalf("a refused swap left a journal row: %+v", jr)
484 }
485 })
486 }
487}
488
489// Review Minor 2: an original the journal's JSON would change (invalid
490// UTF-8 in a free-text column) is refused, so a restore is always exact.
491func TestSwapForImportRefusesAnOriginalJSONWouldChange(t *testing.T) {
492 reg := openTestRegistry(t)
493 old := legacyRow()
494 old.LastError = "bad \xff byte"
495 if err := reg.Upsert(old); err != nil {
496 t.Fatal(err)
497 }
498 stored := mustRoute(t, reg, old.Host)
499 if stored.LastError != old.LastError {
500 t.Skipf("the driver did not keep the raw bytes: %q", stored.LastError)
501 }
502 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
503 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
504 t.Fatalf("swap: %v", err)
505 }
506 if got := mustRoute(t, reg, old.Host); got != stored {
507 t.Fatalf("a refused swap changed the row: %+v", got)
508 }
509}
510
511// Review Minor 6: a failure after the transaction's first write rolls the
512// whole transaction back. A test-only trigger fails the journal write.
513func TestJournalTransactionsAreAtomic(t *testing.T) {
514 reg := openTestRegistry(t)
515 old := legacyRow()
516 if err := reg.Upsert(old); err != nil {
517 t.Fatal(err)
518 }
519 stored := mustRoute(t, reg, old.Host)
520 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
521
522 // The swap: the route DELETE and the exec row INSERT land before the
523 // journal INSERT fails.
524 if _, err := reg.sql.Exec(`CREATE TRIGGER fail_swap BEFORE INSERT ON import_journal BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil {
525 t.Fatal(err)
526 }
527 if err := reg.SwapForImport(j, swapRow()); err == nil {
528 t.Fatal("swap with a failing journal write: no error")
529 }
530 if got := mustRoute(t, reg, old.Host); got != stored {
531 t.Fatalf("a failed swap left %+v, want the original", got)
532 }
533 if jr := mustJournal(t, reg, old.Host); jr != nil {
534 t.Fatalf("a failed swap left a journal row: %+v", jr)
535 }
536 if _, err := reg.sql.Exec(`DROP TRIGGER fail_swap`); err != nil {
537 t.Fatal(err)
538 }
539 if err := reg.SwapForImport(j, swapRow()); err != nil {
540 t.Fatal(err)
541 }
542 if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
543 t.Fatal(err)
544 }
545 closed := mustRoute(t, reg, old.Host)
546
547 // The restore: the route swap lands before the journal UPDATE fails.
548 if _, err := reg.sql.Exec(`CREATE TRIGGER fail_restore BEFORE UPDATE ON import_journal WHEN NEW.state = 'cleanup-pending' BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil {
549 t.Fatal(err)
550 }
551 if err := reg.RestoreImportRoute(old.Host, j.SeedID); err == nil {
552 t.Fatal("restore with a failing journal write: no error")
553 }
554 if got := mustRoute(t, reg, old.Host); got != closed {
555 t.Fatalf("a failed restore left %+v, want the closed exec row", got)
556 }
557 if jr := mustJournal(t, reg, old.Host); jr == nil || jr.State != JournalRollingBack {
558 t.Fatalf("a failed restore left the journal at %+v", jr)
559 }
560 if _, err := reg.sql.Exec(`DROP TRIGGER fail_restore`); err != nil {
561 t.Fatal(err)
562 }
563 if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {
564 t.Fatal(err)
565 }
566 if got := mustRoute(t, reg, old.Host); got != stored {
567 t.Fatalf("restored %+v, want %+v", got, stored)
568 }
569}
570
571// swappedRegistry is a registry where the import has swapped legacyRow for
572// swapRow, and the original as stored.
573func swappedRegistry(t *testing.T) (*Registry, Route) {
574 t.Helper()
575 reg := openTestRegistry(t)
576 if err := reg.Upsert(legacyRow()); err != nil {
577 t.Fatal(err)
578 }
579 stored := mustRoute(t, reg, importHost)
580 j := ImportJournal{Host: importHost, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
581 if err := reg.SwapForImport(j, swapRow()); err != nil {
582 t.Fatal(err)
583 }
584 return reg, stored
585}
586
587// Task 5 (a stale journal row from a snapshot): the rollback's two route
588// transactions act only on the import's own row. A row there that is not
589// it (a later instance at the host, the original put back by hand, a row
590// naming another seed or no longer gated) is refused and left exactly as
591// it is, and the journal does not move.
592func TestRollbackTransactionsRefuseARowThatIsNotTheImports(t *testing.T) {
593 later := execRow()
594 later.ImportGate = ""
595 later.Account = "bbx"
596 for name, foreign := range map[string]Route{
597 "a later instance": later,
598 "the original back": legacyRow(),
599 "another seed's row": func() Route { r := swapRow(); r.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"}`; return r }(),
600 "its gate cleared": func() Route { r := swapRow(); r.ImportGate = ""; return r }(),
601 } {
602 t.Run(name, func(t *testing.T) {
603 for _, closeFirst := range []bool{false, true} {
604 reg, _ := swappedRegistry(t)
605 if closeFirst {
606 if err := reg.CloseImportGate(importHost, importSeedID, 7); err != nil {
607 t.Fatal(err)
608 }
609 }
610 before := mustJournal(t, reg, importHost)
611 if _, err := reg.sql.Exec(`DELETE FROM routes WHERE host = ?`, importHost); err != nil {
612 t.Fatal(err)
613 }
614 if err := reg.Upsert(foreign); err != nil {
615 t.Fatal(err)
616 }
617 placed := mustRoute(t, reg, importHost)
618 if err := reg.CloseImportGate(importHost, importSeedID, 8); !errors.Is(err, ErrImportJournal) {
619 t.Fatalf("close over %s: %v", name, err)
620 }
621 if closeFirst {
622 if err := reg.RestoreImportRoute(importHost, importSeedID); !errors.Is(err, ErrImportJournal) {
623 t.Fatalf("restore over %s: %v", name, err)
624 }
625 }
626 if got := mustRoute(t, reg, importHost); got != placed {
627 t.Fatalf("the row changed: %+v, want %+v", got, placed)
628 }
629 if jr := mustJournal(t, reg, importHost); !reflect.DeepEqual(jr, before) {
630 t.Fatalf("the journal moved: %+v -> %+v", before, jr)
631 }
632 }
633 })
634 }
635}
636
637// Task 5: an import whose row is gone (something removed it) is closed and
638// moved to cleanup-pending with no route put back.
639func TestRollbackTransactionsTolerateAMissingRow(t *testing.T) {
640 reg, _ := swappedRegistry(t)
641 if err := reg.Delete(importHost); err != nil {
642 t.Fatal(err)
643 }
644 if err := reg.CloseImportGate(importHost, importSeedID, 7); err != nil {
645 t.Fatal(err)
646 }
647 if err := reg.RestoreImportRoute(importHost, importSeedID); err != nil {
648 t.Fatal(err)
649 }
650 if rt, err := reg.Get(importHost); err != nil || rt != nil {
651 t.Fatalf("a route came back for a removed row: %+v, %v", rt, err)
652 }
653 if jr := mustJournal(t, reg, importHost); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 7 {
654 t.Fatalf("journal = %+v", jr)
655 }
656}
657
658func TestIsImportRow(t *testing.T) {
659 j := ImportJournal{Host: importHost, SeedID: importSeedID}
660 closed := swapRow()
661 closed.ImportGate = ImportGateClosed
662 if !IsImportRow(swapRow(), j) || !IsImportRow(closed, j) {
663 t.Fatal("the swap's row, gated or closed, is not recognised")
664 }
665 for name, mutate := range map[string]func(*Route){
666 "another host": func(r *Route) { r.Host = "new.shop.example.com" },
667 "a service": func(r *Route) { r.Kind = "service" },
668 "not provisioned": func(r *Route) { r.Provisioned = false },
669 "ungated": func(r *Route) { r.ImportGate = "" },
670 "no seed column": func(r *Route) { r.SeedSource = "" },
671 "a bad seed column": func(r *Route) { r.SeedSource = "{" },
672 "another seed": func(r *Route) { r.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"}` },
673 } {
674 r := swapRow()
675 mutate(&r)
676 if IsImportRow(r, j) {
677 t.Errorf("%s: recognised as the import's row", name)
678 }
679 }
680 if IsImportRow(swapRow(), ImportJournal{Host: importHost}) {
681 t.Error("a journal with no seed id recognised a row")
682 }
683}
684
685// Fix round 1, I2: a delete decided from an earlier list removes the row
686// only while it is still the row listed. A rollback that put the original
687// (another owner's, with its grants) back in between keeps it.
688func TestDeleteIfUnchanged(t *testing.T) {
689 reg, stored := swappedRegistry(t)
690 listed := mustRoute(t, reg, importHost)
691 if err := reg.CloseImportGate(importHost, importSeedID, 7); err != nil {
692 t.Fatal(err)
693 }
694 closed := mustRoute(t, reg, importHost)
695 if err := reg.RestoreImportRoute(importHost, importSeedID); err != nil {
696 t.Fatal(err)
697 }
698 for name, stale := range map[string]Route{"the gated import row": listed, "the closed import row": closed} {
699 if ok, err := reg.DeleteIfUnchanged(stale); err != nil || ok {
700 t.Fatalf("%s: deleted %v err %v", name, ok, err)
701 }
702 }
703 if got := mustRoute(t, reg, importHost); got != stored {
704 t.Fatalf("the restored row changed: %+v", got)
705 }
706 for name, mutate := range map[string]func(*Route){
707 "another account": func(r *Route) { r.Account = "bbw" },
708 "another app": func(r *Route) { r.App = "other" },
709 "another kind": func(r *Route) { r.Kind = "instance" },
710 "provisioned": func(r *Route) { r.Provisioned = true },
711 "a gate": func(r *Route) { r.ImportGate = ImportGateClosed },
712 "a seed column": func(r *Route) { r.SeedSource = `{"id":"` + importSeedID + `"}` },
713 "another creation": func(r *Route) { r.CreatedAt++ },
714 } {
715 stale := stored
716 mutate(&stale)
717 if ok, err := reg.DeleteIfUnchanged(stale); err != nil || ok {
718 t.Fatalf("%s: deleted %v err %v", name, ok, err)
719 }
720 }
721 if ok, err := reg.DeleteIfUnchanged(stored); err != nil || !ok {
722 t.Fatalf("the row as listed: deleted %v err %v", ok, err)
723 }
724 if rt, _ := reg.Get(importHost); rt != nil {
725 t.Fatalf("still there: %+v", rt)
726 }
727 if ok, err := reg.DeleteIfUnchanged(stored); err != nil || ok {
728 t.Fatalf("a second delete: %v %v", ok, err)
729 }
730}
731
732// The merge of the seeded columns with main's good-build, serving, settings
733// and build-state columns: a registry written by either side alone gains
734// the other side's columns and tables on open, every time, keeps the row it
735// held, and then round-trips a row that sets all of them.
736func TestARegistryFromEitherSideGainsTheOthersColumns(t *testing.T) {
737 seeded := []string{"seed_source", "import_gate"}
738 mains := []string{"good_version", "good_digest", "good_demand", "serving_version", "update_settings", "build_state"}
739 for _, tc := range []struct {
740 name string
741 drop []string // the columns the side that wrote the file never had
742 tables []string // and its tables
743 }{
744 {"a seeded-side file", mains, nil},
745 {"a main-side file", seeded, []string{"import_journal", "fence_generation"}},
746 } {
747 t.Run(tc.name, func(t *testing.T) {
748 path := filepath.Join(t.TempDir(), "registry.db")
749 r, err := Open(path)
750 if err != nil {
751 t.Fatal(err)
752 }
753 if err := r.Upsert(legacyRow()); err != nil {
754 t.Fatal(err)
755 }
756 kept := mustRoute(t, r, importHost)
757 for _, c := range tc.drop {
758 if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN ` + c); err != nil {
759 t.Fatal(err)
760 }
761 }
762 for _, tb := range tc.tables {
763 if _, err := r.sql.Exec(`DROP TABLE ` + tb); err != nil {
764 t.Fatal(err)
765 }
766 }
767 r.Close()
768 for i := 0; i < 2; i++ {
769 r, err = Open(path)
770 if err != nil {
771 t.Fatalf("open %d: %v", i, err)
772 }
773 if got := mustRoute(t, r, importHost); got != kept {
774 t.Fatalf("open %d: the row it held reads %+v, want %+v", i, got, kept)
775 }
776 if js, err := r.ImportJournals(); err != nil || len(js) != 0 {
777 t.Fatalf("open %d: journals %+v %v", i, js, err)
778 }
779 if g, err := r.NextFenceGeneration(time.Unix(1, 0)); err != nil || g == 0 {
780 t.Fatalf("open %d: fence generation %d, %v", i, g, err)
781 }
782 r.Close()
783 }
784 r, err = Open(path)
785 if err != nil {
786 t.Fatal(err)
787 }
788 defer r.Close()
789 want := swapRow()
790 want.Host = "both.shop.example.com"
791 want.GoodVersion, want.GoodDigest, want.GoodDemand = "good789", "sha256:good", "swap-demand"
792 want.ServingVersion = "serve012"
793 want.UpdateSettings = `{"updates":"manual","channel":"stable"}`
794 want.BuildState = `{"known":true,"state":{"schema":1,"last_executed":"bs345"}}`
795 if want.SeedSource == "" || want.ImportGate == "" {
796 t.Fatal("the fixture's exec row no longer sets the seeded columns")
797 }
798 if err := r.Upsert(want); err != nil {
799 t.Fatal(err)
800 }
801 got := mustRoute(t, r, want.Host)
802 want.CreatedAt, want.Status, want.Kind = got.CreatedAt, got.Status, got.Kind
803 if got != want {
804 t.Fatalf("round trip after the migration:\n got %+v\n want %+v", got, want)
805 }
806 })
807 }
808}
809
810// Merge review S1: the journal keeps what the host's own env file held
811// before the import (EnvBefore), through every state, so the rollback can
812// tell the key the import minted from one that predates it.
813func TestImportJournalKeepsEnvBefore(t *testing.T) {
814 for _, before := range []string{"", EnvBeforeAbsent, EnvBeforeNoKey, EnvBeforeKey} {
815 reg := openTestRegistry(t)
816 old := legacyRow()
817 if err := reg.Upsert(old); err != nil {
818 t.Fatal(err)
819 }
820 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: mustRoute(t, reg, old.Host), At: 1, EnvBefore: before}
821 if err := reg.SwapForImport(j, swapRow()); err != nil {
822 t.Fatalf("%q: %v", before, err)
823 }
824 if got := mustJournal(t, reg, old.Host); got == nil || got.EnvBefore != before {
825 t.Fatalf("after the swap: %+v, want env before %q", got, before)
826 }
827 if err := reg.CloseImportGate(old.Host, j.SeedID, 7); err != nil {
828 t.Fatal(err)
829 }
830 if got := mustJournal(t, reg, old.Host); got.EnvBefore != before || got.State != JournalRollingBack {
831 t.Fatalf("rolling back: %+v, want env before %q", got, before)
832 }
833 if js, err := reg.ImportJournals(); err != nil || len(js) != 1 || js[0].EnvBefore != before {
834 t.Fatalf("ImportJournals = %+v, %v", js, err)
835 }
836 }
837 reg := openTestRegistry(t)
838 old := legacyRow()
839 if err := reg.Upsert(old); err != nil {
840 t.Fatal(err)
841 }
842 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: mustRoute(t, reg, old.Host), At: 1, EnvBefore: "maybe"}
843 if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
844 t.Fatalf("a swap with an env state that is none of the recorded ones: %v", err)
845 }
846}
847
848// A journal table from before env_before gains the column on open, and its
849// rows read as "nothing recorded". Such a row digests as it did; a
850// recorded value moves the snapshot digest.
851func TestImportJournalEnvBeforeMigratesAndIsHashedWhenSet(t *testing.T) {
852 path := filepath.Join(t.TempDir(), "registry.db")
853 r, err := Open(path)
854 if err != nil {
855 t.Fatal(err)
856 }
857 old := legacyRow()
858 if err := r.Upsert(old); err != nil {
859 t.Fatal(err)
860 }
861 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: mustRoute(t, r, old.Host), At: 1}
862 if err := r.SwapForImport(j, swapRow()); err != nil {
863 t.Fatal(err)
864 }
865 unset := mustDigest(t, r.SnapshotFingerprint)
866 if _, err := r.sql.Exec(`ALTER TABLE import_journal DROP COLUMN env_before`); err != nil {
867 t.Fatal(err)
868 }
869 r.Close()
870 for i := 0; i < 2; i++ {
871 if r, err = Open(path); err != nil {
872 t.Fatalf("open %d: %v", i, err)
873 }
874 if got := mustJournal(t, r, old.Host); got == nil || got.EnvBefore != "" || got.SeedID != importSeedID {
875 t.Fatalf("open %d: journal %+v", i, got)
876 }
877 if got := mustDigest(t, r.SnapshotFingerprint); got != unset {
878 t.Fatalf("open %d: a row with nothing recorded digests %s, was %s", i, got, unset)
879 }
880 if i == 0 {
881 r.Close()
882 }
883 }
884 defer r.Close()
885 if _, err := r.sql.Exec(`UPDATE import_journal SET env_before = ? WHERE host = ?`, EnvBeforeAbsent, old.Host); err != nil {
886 t.Fatal(err)
887 }
888 if mustDigest(t, r.SnapshotFingerprint) == unset {
889 t.Fatal("a recorded env state did not move SnapshotFingerprint")
890 }
891}
892
893// Merge review S2: a box says which hosts its import journal holds, for the
894// claim activation census. The read answers through root's registry child
895// too, where the journal's other methods refuse, so a collector that runs
896// as root is not left unable to say.
897func TestImportJournalHostsLocalAndRemote(t *testing.T) {
898 r, path := remoteTestRegistry(t)
899 if hosts, err := r.ImportJournalHosts(); err != nil || hosts == nil || len(hosts) != 0 {
900 t.Fatalf("an empty journal through the child = %v, %v; want an empty list", hosts, err)
901 }
902 local, err := Open(path)
903 if err != nil {
904 t.Fatal(err)
905 }
906 defer local.Close()
907 old := legacyRow()
908 if err := local.Upsert(old); err != nil {
909 t.Fatal(err)
910 }
911 j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: mustRoute(t, local, old.Host), At: 1}
912 if err := local.SwapForImport(j, swapRow()); err != nil {
913 t.Fatal(err)
914 }
915 for name, reg := range map[string]*Registry{"local": local, "through the child": r} {
916 hosts, err := reg.ImportJournalHosts()
917 if err != nil || len(hosts) != 1 || hosts[0] != old.Host {
918 t.Fatalf("%s: ImportJournalHosts = %v, %v", name, hosts, err)
919 }
920 }
921 // The rest of the journal stays the agent's own.
922 if _, err := r.ImportJournals(); !errors.Is(err, errRemoteMethod) {
923 t.Fatalf("ImportJournals through the child = %v, want the remote refusal", err)
924 }
925}
926