amadan

carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1package provision
2
3import (
4 "context"
5 "database/sql"
6 "encoding/json"
7 "errors"
8 "os"
9 "reflect"
10 "strings"
11 "testing"
12
13 "github.com/carlosframework/platform/internal/registry"
14 "github.com/carlosframework/platform/internal/release"
15 "github.com/carlosframework/platform/internal/seeds"
16)
17
18// rollbackRecorder is the brief's fake: it records the box effects in the
19// order they happen.
20type rollbackRecorder struct {
21 steps []string
22 proceed bool
23 fenceErr error
24 gen uint64
25 marker string // the seed id the local database's marker names
26 // retired is each RetireKey call, "<db path> <host> <env before>". It
27 // is kept apart from steps, whose sequences predate the hook.
28 retired []string
29}
30
31func (r *rollbackRecorder) hooks(h *SeedHooks) {
32 h.NextFenceGen = func() (uint64, error) { r.gen++; return r.gen, nil }
33 h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) {
34 r.steps = append(r.steps, "fence")
35 return r.proceed, r.fenceErr
36 }
37 h.LiftFence = func(host string, gen uint64) { r.steps = append(r.steps, "lift") }
38 h.RemoveLocal = func(rt registry.Route) error { r.steps = append(r.steps, "remove-local:"+rt.Kind); return nil }
39 h.DeleteReplica = func(_ context.Context, host string) error { r.steps = append(r.steps, "delete-replica"); return nil }
40 h.LocalSeedID = func(string) (string, error) { return r.marker, nil }
41 h.RetireKey = func(db, host, before string) error {
42 r.retired = append(r.retired, db+" "+host+" "+before)
43 return nil
44 }
45}
46
47func TestRollbackAfterAbortRestoresTheRowExactly(t *testing.T) {
48 s, sqid := shopSetup(t)
49 rec := importRecord(t, s, sqid)
50 reg := regWith(t, addrRow(rec, sqid))
51 original, _ := reg.Get(rec.Host)
52 cfg := importCfg(t, nil)
53 rr := &rollbackRecorder{proceed: true}
54 rr.hooks(cfg.Seeds)
55 cfg.Seeds.Data.Put(ctx, seeds.SnapshotKey(rec.Host, rec.Seed.ID), []byte("db"), "")
56 other, _ := seeds.NewID()
57 cfg.Seeds.Data.Put(ctx, seeds.SnapshotKey(rec.Host, other), []byte("db"), "")
58 Reconcile(ctx, s, reg, cfg) // swap
59 if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
60 t.Fatal(err)
61 }
62 if err := Reconcile(ctx, s, reg, cfg); err != nil {
63 t.Fatal(err)
64 }
65 if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) {
66 t.Fatalf("restored %+v, want %+v", got, original)
67 }
68 if j, _ := reg.ImportJournal(rec.Host); j != nil {
69 t.Fatalf("journal = %+v", j)
70 }
71 want := []string{"fence", "remove-local:instance", "lift", "delete-replica"}
72 if !reflect.DeepEqual(rr.steps, want) {
73 t.Fatalf("steps = %q, want %q", rr.steps, want)
74 }
75 if _, _, err := cfg.Seeds.Data.Get(ctx, seeds.SnapshotKey(rec.Host, rec.Seed.ID)); !errors.Is(err, os.ErrNotExist) {
76 t.Fatal("this seed's snapshot survived")
77 }
78 if _, _, err := cfg.Seeds.Data.Get(ctx, seeds.SnapshotKey(rec.Host, other)); err != nil {
79 t.Fatal("another seed's snapshot was deleted")
80 }
81 r, _ := GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
82 if r.Seed.State != seeds.StateRolledBack {
83 t.Fatalf("record = %s", r.Seed.State)
84 }
85}
86
87func TestRollbackForAGoneOrReplacedRecord(t *testing.T) {
88 for _, mutate := range []string{"deleted", "another seed"} {
89 t.Run(mutate, func(t *testing.T) {
90 s, sqid := shopSetup(t)
91 rec := importRecord(t, s, sqid)
92 reg := regWith(t, addrRow(rec, sqid))
93 original, _ := reg.Get(rec.Host)
94 cfg := importCfg(t, nil)
95 rr := &rollbackRecorder{proceed: true}
96 rr.hooks(cfg.Seeds)
97 Reconcile(ctx, s, reg, cfg)
98 if err := DeleteRecord(ctx, s, rec.Account, rec.App, rec.Host); err != nil {
99 t.Fatal(err)
100 }
101 var next Record
102 if mutate == "another seed" {
103 next = importRecord(t, s, sqid) // the same host, a new seed id, state ready
104 }
105 if err := Reconcile(ctx, s, reg, cfg); err != nil {
106 t.Fatal(err)
107 }
108 if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) {
109 t.Fatalf("restored %+v", got)
110 }
111 if j, _ := reg.ImportJournal(rec.Host); j != nil {
112 t.Fatalf("journal = %+v", j)
113 }
114 // A record naming another seed holds the host: the host's
115 // replica is not this rollback's to delete, and that record
116 // keeps its own state.
117 wantSteps := []string{"fence", "remove-local:instance", "lift", "delete-replica"}
118 if mutate == "another seed" {
119 wantSteps = wantSteps[:3]
120 if st := seedState(t, s, next); st.ID != next.Seed.ID || st.State != seeds.StateReady {
121 t.Fatalf("the new record's seed = %+v", st)
122 }
123 }
124 if !reflect.DeepEqual(rr.steps, wantSteps) {
125 t.Fatalf("steps = %q, want %q", rr.steps, wantSteps)
126 }
127 })
128 }
129}
130
131func TestRollbackDefersWhenTheFenceIsNotOurs(t *testing.T) {
132 s, sqid := shopSetup(t)
133 rec := importRecord(t, s, sqid)
134 reg := regWith(t, addrRow(rec, sqid))
135 cfg := importCfg(t, nil)
136 rr := &rollbackRecorder{proceed: false}
137 rr.hooks(cfg.Seeds)
138 Reconcile(ctx, s, reg, cfg)
139 SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID))
140 Reconcile(ctx, s, reg, cfg)
141 j, _ := reg.ImportJournal(rec.Host)
142 got := mustGet(t, reg, rec.Host)
143 if j == nil || j.State != registry.JournalRollingBack || got.ImportGate != registry.ImportGateClosed || got.Kind != "instance" {
144 t.Fatalf("after a deferred fence: journal %+v row %+v; want rolling-back, gate closed, row kept", j, got)
145 }
146 if !reflect.DeepEqual(rr.steps, []string{"fence"}) {
147 t.Fatalf("steps = %q: a deferred fence went on to remove or delete", rr.steps)
148 }
149 // Next pass the fence proceeds, with the SAME generation.
150 gen := j.FenceGen
151 rr.proceed = true
152 Reconcile(ctx, s, reg, cfg)
153 if rr.gen != gen {
154 t.Fatalf("a resumed rollback minted a new generation: %d, journal had %d", rr.gen, gen)
155 }
156 if j, _ := reg.ImportJournal(rec.Host); j != nil {
157 t.Fatalf("journal = %+v", j)
158 }
159}
160
161// A fence hook that answers with an error defers, whatever proceed says,
162// and this rollback's generation is lifted (review Minor 1).
163func TestRollbackDefersOnAFenceError(t *testing.T) {
164 x := abortedImport(t)
165 rr := &rollbackRecorder{proceed: true, fenceErr: errors.New("stop unproven")}
166 rr.hooks(x.cfg.Seeds)
167 x.pass(t)
168 if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || j.State != registry.JournalRollingBack {
169 t.Fatalf("journal = %+v", j)
170 }
171 if !reflect.DeepEqual(rr.steps, []string{"fence", "lift"}) {
172 t.Fatalf("steps = %q", rr.steps)
173 }
174}
175
176func TestCleanupPendingNeverFences(t *testing.T) {
177 s, sqid := shopSetup(t)
178 rec := importRecord(t, s, sqid)
179 reg := regWith(t, addrRow(rec, sqid))
180 cfg := importCfg(t, nil)
181 rr := &rollbackRecorder{proceed: true}
182 rr.hooks(cfg.Seeds)
183 Reconcile(ctx, s, reg, cfg)
184 SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID))
185 // Crash after the route restore, before the bucket cleanup.
186 if err := reg.CloseImportGate(rec.Host, rec.Seed.ID, 9); err != nil {
187 t.Fatal(err)
188 }
189 if err := reg.RestoreImportRoute(rec.Host, rec.Seed.ID); err != nil {
190 t.Fatal(err)
191 }
192 rr.steps = nil
193 if err := Reconcile(ctx, s, reg, cfg); err != nil {
194 t.Fatal(err)
195 }
196 if !reflect.DeepEqual(rr.steps, []string{"delete-replica"}) {
197 t.Fatalf("steps = %q; a cleanup-pending resume fenced, stopped or removed something", rr.steps)
198 }
199 if rr.gen != 0 {
200 t.Fatalf("a cleanup-pending resume minted a fence generation")
201 }
202}
203
204func TestJournalDoesNotRollBackOnAReadFailure(t *testing.T) {
205 for _, how := range []string{"unparseable body", "read error"} {
206 t.Run(how, func(t *testing.T) {
207 s, sqid := shopSetup(t)
208 rec := importRecord(t, s, sqid)
209 reg := regWith(t, addrRow(rec, sqid))
210 cfg := importCfg(t, nil)
211 rr := &rollbackRecorder{proceed: true}
212 rr.hooks(cfg.Seeds)
213 Reconcile(ctx, s, reg, cfg)
214 var store release.Store = s
215 key := RecordKey(rec.Account, rec.App, rec.Host)
216 if how == "unparseable body" {
217 // The record body becomes unreadable: protected, never "gone".
218 putReplace(t, s, key, []byte("{not json"))
219 } else {
220 // preflight m7: a Get that fails (readFailed), not a bad body.
221 store = getFailsFor{s, key}
222 }
223 if err := Reconcile(ctx, store, reg, cfg); err != nil {
224 t.Fatal(err)
225 }
226 if len(rr.steps) != 0 || rr.gen != 0 {
227 t.Fatalf("an unreadable record started a rollback: steps %q", rr.steps)
228 }
229 if j, _ := reg.ImportJournal(rec.Host); j == nil || j.State != registry.JournalSwapped {
230 t.Fatalf("journal = %+v", j)
231 }
232 if got := mustGet(t, reg, rec.Host); got.Kind != "instance" {
233 t.Fatal("the removal phase touched a journaled host")
234 }
235 })
236 }
237}
238
239// getFailsFor fails every read of one key with something other than "not
240// found".
241type getFailsFor struct {
242 release.Store
243 key string
244}
245
246func (g getFailsFor) Get(ctx context.Context, key string) ([]byte, string, error) {
247 if key == g.key {
248 return nil, "", errors.New("connection reset")
249 }
250 return g.Store.Get(ctx, key)
251}
252
253// ---- the box side, with the step order checked as it happens -------------
254
255// crashed is the panic a fault raises to kill the "process" mid-pass.
256type crashed struct{}
257
258// rollbackBox is the box's side of the rollback: an in-memory fence (lost
259// when the process dies), the import's local data, and the host's replica.
260// Every hook checks, when it is called, the order the spec requires: the
261// fence comes after the gate is closed and under the journal's own
262// generation; nothing local is removed unless the host is fenced and the
263// journal is still rolling back; nothing in the bucket is deleted before
264// the original route is back.
265type rollbackBox struct {
266 t *testing.T
267 reg *registry.Registry
268 original registry.Route
269 steps []string
270 proceed bool
271 gen uint64
272 fenced map[string]uint64
273 local bool
274 replica bool
275 fault string // the step that fails, once
276 crash bool // the fault kills the process (a panic) rather than returning an error
277 noRoute bool // the import's row was removed: no route comes back
278 marker string // the seed id the local database's <db>.seed names
279 markErr error // reading that marker fails
280 // retired is each RetireKey call's env-before value, kept apart from
281 // steps; key is whether the import's minted key is still on the box.
282 retired []string
283 key bool
284}
285
286func newRollbackBox(t *testing.T, reg *registry.Registry, original registry.Route) *rollbackBox {
287 return &rollbackBox{t: t, reg: reg, original: original, proceed: true, fenced: map[string]uint64{}, local: true, replica: true, key: true}
288}
289
290func (b *rollbackBox) trip(step string) error {
291 if b.fault != step {
292 return nil
293 }
294 b.fault = ""
295 if b.crash {
296 panic(crashed{})
297 }
298 return errors.New("injected failure at " + step)
299}
300
301// restart is the process dying: the activator's fences are in memory.
302func (b *rollbackBox) restart() { b.fenced = map[string]uint64{} }
303
304func (b *rollbackBox) journal() *registry.ImportJournal {
305 j, err := b.reg.ImportJournal(b.original.Host)
306 if err != nil {
307 b.t.Fatal(err)
308 }
309 return j
310}
311
312func (b *rollbackBox) row() *registry.Route {
313 rt, err := b.reg.Get(b.original.Host)
314 if err != nil {
315 b.t.Fatal(err)
316 }
317 return rt
318}
319
320// wantRestored: nothing is deleted from the bucket before the original
321// route is back and the journal says cleanup-pending (or is gone, for the
322// no-journal finish).
323func (b *rollbackBox) wantRestored(what string) {
324 switch rt := b.row(); {
325 case b.noRoute:
326 if rt != nil {
327 b.t.Errorf("%s with a route at the host: %+v", what, rt)
328 }
329 case rt == nil || !reflect.DeepEqual(*rt, b.original):
330 b.t.Errorf("%s before the original route was back: row %+v", what, rt)
331 }
332 if j := b.journal(); j != nil && j.State != registry.JournalCleanupPending {
333 b.t.Errorf("%s with the journal at %s", what, j.State)
334 }
335 if len(b.fenced) != 0 {
336 b.t.Errorf("%s with a fence held: the cleanup never fences", what)
337 }
338}
339
340func (b *rollbackBox) hooks(h *SeedHooks) {
341 h.NextFenceGen = func() (uint64, error) {
342 b.steps = append(b.steps, "gen")
343 b.gen++
344 return b.gen, nil
345 }
346 h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) {
347 b.steps = append(b.steps, "fence")
348 if j := b.journal(); j == nil || j.State != registry.JournalRollingBack || j.FenceGen != gen {
349 b.t.Errorf("fenced with generation %d under journal %+v", gen, j)
350 }
351 if rt := b.row(); rt != nil && rt.ImportGate != registry.ImportGateClosed {
352 b.t.Errorf("fenced before the gate was closed: %+v", rt)
353 }
354 if err := b.trip("fence"); err != nil {
355 return false, err
356 }
357 if held, ok := b.fenced[host]; ok {
358 return held == gen, nil // ours from an earlier pass, or another operation's
359 }
360 if !b.proceed {
361 return false, nil
362 }
363 b.fenced[host] = gen
364 return true, nil
365 }
366 h.LiftFence = func(host string, gen uint64) {
367 b.steps = append(b.steps, "lift")
368 _ = b.trip("lift") // a lift cannot fail; only a crash stops it
369 if b.fenced[host] == gen {
370 delete(b.fenced, host)
371 }
372 }
373 h.RemoveLocal = func(rt registry.Route) error {
374 b.steps = append(b.steps, "remove-local")
375 if _, ok := b.fenced[rt.Host]; !ok {
376 b.t.Errorf("removed local data of a host that is not fenced and stopped")
377 }
378 if j := b.journal(); j == nil || j.State != registry.JournalRollingBack {
379 b.t.Errorf("removed local data with the journal at %+v", j)
380 }
381 // The closed import row, or (its row gone) the host's default
382 // nested database path.
383 if rt.DBPath != NestedDBPath("/data", rt.Host) || (rt.ImportGate != registry.ImportGateClosed && !b.noRoute) {
384 b.t.Errorf("removed local data for a row that is not the closed import row: %+v", rt)
385 }
386 if err := b.trip("remove-local"); err != nil {
387 return err
388 }
389 b.local = false
390 return b.trip("remove-local-done")
391 }
392 h.RetireKey = func(dbPath, host, before string) error {
393 b.retired = append(b.retired, before)
394 if _, ok := b.fenced[host]; !ok {
395 b.t.Errorf("retired the instance key of a host that is not fenced and stopped")
396 }
397 if j := b.journal(); j == nil || j.State != registry.JournalRollingBack || j.EnvBefore != before {
398 b.t.Errorf("retired the instance key by %q under journal %+v", before, j)
399 }
400 if dbPath != NestedDBPath("/data", host) {
401 b.t.Errorf("retired the instance key beside %s", dbPath)
402 }
403 // Before the local data: its marker is the missing-row path's
404 // evidence that the installation is the import's. (A resumed pass
405 // retires again after the data went; that is a no-op.)
406 if b.key && !b.local {
407 b.t.Errorf("the instance key was first retired after the local data was removed")
408 }
409 if err := b.trip("retire-key"); err != nil {
410 return err
411 }
412 b.key = false
413 return nil
414 }
415 h.LocalSeedID = func(dbPath string) (string, error) {
416 b.steps = append(b.steps, "read-marker")
417 if dbPath != NestedDBPath("/data", b.original.Host) {
418 b.t.Errorf("read the marker beside %s", dbPath)
419 }
420 if b.markErr != nil {
421 return "", b.markErr
422 }
423 if !b.local {
424 return "", nil
425 }
426 return b.marker, nil
427 }
428 h.DeleteReplica = func(_ context.Context, host string) error {
429 b.steps = append(b.steps, "delete-replica")
430 b.wantRestored("deleted the replica")
431 if err := b.trip("delete-replica"); err != nil {
432 return err
433 }
434 b.replica = false
435 return nil
436 }
437}
438
439// faultData is the region data bucket, with the snapshot delete checked and
440// faultable.
441type faultData struct {
442 release.Store
443 box *rollbackBox
444}
445
446func (d faultData) Delete(ctx context.Context, key string) error {
447 if strings.HasPrefix(key, seeds.Prefix) {
448 d.box.steps = append(d.box.steps, "delete-snapshot")
449 d.box.wantRestored("deleted the snapshot")
450 if err := d.box.trip("delete-snapshot"); err != nil {
451 return err
452 }
453 }
454 return d.Store.Delete(ctx, key)
455}
456
457// faultRecords is the release bucket, with the record CAS faultable.
458type faultRecords struct {
459 release.Store
460 box *rollbackBox
461 key string
462}
463
464func (f faultRecords) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
465 if key == f.key {
466 if err := f.box.trip("cas"); err != nil {
467 return "", err
468 }
469 }
470 return f.Store.Put(ctx, key, body, ifMatch)
471}
472
473// rollbackScene is an import this box swapped, verified once (its local
474// database placed and replicated) and then aborted.
475type rollbackScene struct {
476 raw *release.DirStore
477 store release.Store
478 reg *registry.Registry
479 path string
480 rec Record
481 other string // another seed of the host, with its own snapshot
482 cfg ReconcileConfig
483 box *rollbackBox
484}
485
486func abortedImport(t *testing.T) *rollbackScene {
487 t.Helper()
488 s, sqid := shopSetup(t)
489 rec := importRecord(t, s, sqid)
490 reg, path := regAt(t, addrRow(rec, sqid))
491 box := newRollbackBox(t, reg, *mustGet(t, reg, rec.Host))
492 box.marker = rec.Seed.ID // the verify wake placed this seed
493 cfg := importCfg(t, nil)
494 box.hooks(cfg.Seeds)
495 data := cfg.Seeds.Data
496 cfg.Seeds.Data = faultData{data, box}
497 other, _ := seeds.NewID()
498 for _, id := range []string{rec.Seed.ID, other} {
499 if _, err := data.Put(ctx, seeds.SnapshotKey(rec.Host, id), []byte("db"), ""); err != nil {
500 t.Fatal(err)
501 }
502 }
503 x := &rollbackScene{raw: s, store: faultRecords{s, box, RecordKey(rec.Account, rec.App, rec.Host)},
504 reg: reg, path: path, rec: rec, other: other, cfg: cfg, box: box}
505 x.pass(t)
506 if got := mustGet(t, reg, rec.Host); !registry.IsImportRow(*got, registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID}) {
507 t.Fatalf("the swap did not happen: %+v", got)
508 }
509 if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
510 t.Fatal(err)
511 }
512 box.steps = nil
513 return x
514}
515
516// pass is one Reconcile. A fault that crashes the process ends the pass
517// where it stands, and the box restarts.
518func (x *rollbackScene) pass(t *testing.T) {
519 t.Helper()
520 defer func() {
521 if r := recover(); r != nil {
522 if _, ok := r.(crashed); !ok {
523 panic(r)
524 }
525 x.box.restart()
526 }
527 }()
528 if err := Reconcile(ctx, x.store, x.reg, x.cfg); err != nil {
529 t.Fatal(err)
530 }
531}
532
533func (x *rollbackScene) journalState(t *testing.T) string {
534 t.Helper()
535 j, err := x.reg.ImportJournal(x.rec.Host)
536 if err != nil {
537 t.Fatal(err)
538 }
539 if j == nil {
540 return ""
541 }
542 return j.State
543}
544
545// wantRolledBack is the clean end state: the original route exactly, no
546// journal, this seed's snapshot gone and the other kept, no local data, no
547// replica, no fence, and the record rolled back.
548func (x *rollbackScene) wantRolledBack(t *testing.T) {
549 t.Helper()
550 if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) {
551 t.Fatalf("row = %+v, want the original %+v", got, x.box.original)
552 }
553 if st := x.journalState(t); st != "" {
554 t.Fatalf("journal still at %s", st)
555 }
556 data := x.cfg.Seeds.Data
557 if _, _, err := data.Get(ctx, seeds.SnapshotKey(x.rec.Host, x.rec.Seed.ID)); !errors.Is(err, os.ErrNotExist) {
558 t.Fatalf("this seed's snapshot: %v", err)
559 }
560 if _, _, err := data.Get(ctx, seeds.SnapshotKey(x.rec.Host, x.other)); err != nil {
561 t.Fatalf("another seed's snapshot: %v", err)
562 }
563 if x.box.local || x.box.replica || len(x.box.fenced) != 0 {
564 t.Fatalf("local %v replica %v fenced %v", x.box.local, x.box.replica, x.box.fenced)
565 }
566 // The key the import's verify minted is retired, by what the swap
567 // recorded: this host had no env file before the import (importCfg).
568 if x.box.key || len(x.box.retired) == 0 {
569 t.Fatalf("the import's instance key was not retired (calls %q)", x.box.retired)
570 }
571 for _, before := range x.box.retired {
572 if before != registry.EnvBeforeAbsent {
573 t.Fatalf("retired by %q, want what the swap recorded (%q)", before, registry.EnvBeforeAbsent)
574 }
575 }
576 if st := seedState(t, x.raw, x.rec); st.ID != x.rec.Seed.ID || st.State != seeds.StateRolledBack {
577 t.Fatalf("record seed = %+v, want rolled-back", st)
578 }
579}
580
581func (x *rollbackScene) trigger(t *testing.T, ddl string) {
582 t.Helper()
583 addTrigger(t, x.path, ddl)
584}
585
586func count(steps []string, step string) int {
587 n := 0
588 for _, s := range steps {
589 if s == step {
590 n++
591 }
592 }
593 return n
594}
595
596// Spec §9 and preflight M13: a failure between any two of the rollback's
597// steps, as an error the process survives or as a crash that loses the
598// in-memory fence, is resumed by the next passes to the same clean end,
599// with nothing lost and nothing done twice that must not be: one fence
600// generation per close, no local removal or fence once the route is back,
601// and no bucket delete before it is.
602//
603// The three registry-trigger rows "crash" as a failed transaction followed
604// by a restart, not as a panic inside the transaction: SQLite rolls the
605// transaction back either way, so the state left is the same.
606func TestRollbackResumesAfterAFailureBetweenAnyTwoSteps(t *testing.T) {
607 const raise = ` BEGIN SELECT RAISE(ABORT, 'injected'); END`
608 for _, f := range []struct {
609 name string
610 hook string // a box hook or a bucket write that fails once
611 trigger string // or a registry write that fails, until dropped
612 after string // the journal's state after the failed pass
613 crashOnly bool
614 }{
615 {name: "closing the gate", trigger: `CREATE TRIGGER fault BEFORE UPDATE OF state ON import_journal WHEN NEW.state = 'rolling-back'` + raise, after: registry.JournalSwapped},
616 {name: "fencing", hook: "fence", after: registry.JournalRollingBack},
617 {name: "retiring the instance key", hook: "retire-key", after: registry.JournalRollingBack},
618 {name: "removing the local data", hook: "remove-local", after: registry.JournalRollingBack},
619 {name: "after removing the local data", hook: "remove-local-done", after: registry.JournalRollingBack},
620 {name: "restoring the route", trigger: `CREATE TRIGGER fault BEFORE UPDATE OF state ON import_journal WHEN NEW.state = 'cleanup-pending'` + raise, after: registry.JournalRollingBack},
621 {name: "lifting the fence", hook: "lift", after: registry.JournalCleanupPending, crashOnly: true},
622 {name: "deleting the snapshot", hook: "delete-snapshot", after: registry.JournalCleanupPending},
623 {name: "deleting the replica", hook: "delete-replica", after: registry.JournalCleanupPending},
624 {name: "deleting the journal row", trigger: `CREATE TRIGGER fault BEFORE DELETE ON import_journal` + raise, after: registry.JournalCleanupPending},
625 {name: "marking the record rolled back", hook: "cas", after: ""},
626 } {
627 for _, crash := range []bool{false, true} {
628 if f.crashOnly && !crash {
629 continue
630 }
631 name := f.name + "/error"
632 if crash {
633 name = f.name + "/crash"
634 }
635 t.Run(name, func(t *testing.T) {
636 x := abortedImport(t)
637 x.box.fault, x.box.crash = f.hook, crash
638 if f.trigger != "" {
639 x.trigger(t, f.trigger)
640 }
641 x.pass(t)
642 if x.box.fault != "" {
643 t.Fatalf("the fault at %s was never reached: steps %q", f.hook, x.box.steps)
644 }
645 if st := x.journalState(t); st != f.after {
646 t.Fatalf("after the failed pass the journal is %q, want %q", st, f.after)
647 }
648 if f.trigger != "" {
649 x.trigger(t, `DROP TRIGGER fault`)
650 if crash {
651 x.box.restart()
652 }
653 }
654 if f.after == "" {
655 if st := seedState(t, x.raw, x.rec); st.State != seeds.StateAborted {
656 t.Fatalf("record = %s, want aborted with no journal row", st.State)
657 }
658 }
659 restoredAt := -1
660 for i := 0; i < 3; i++ {
661 if restoredAt < 0 && x.journalState(t) != registry.JournalSwapped && x.journalState(t) != registry.JournalRollingBack {
662 restoredAt = len(x.box.steps)
663 }
664 x.pass(t)
665 }
666 x.wantRolledBack(t)
667 gens := 1
668 if f.after == registry.JournalSwapped {
669 gens = 2 // the failed close's generation was never stored
670 }
671 if n := count(x.box.steps, "gen"); n != gens {
672 t.Fatalf("minted %d fence generations, want %d: %q", n, gens, x.box.steps)
673 }
674 if restoredAt >= 0 {
675 for _, s := range x.box.steps[restoredAt:] {
676 if s == "fence" || s == "remove-local" || s == "gen" {
677 t.Fatalf("%s after the route was restored: %q", s, x.box.steps)
678 }
679 }
680 }
681 })
682 }
683 }
684}
685
686// Preflight M13: the record is deleted and the fence defers. The gated row
687// is Provisioned and its record gone, which is what the removal phase
688// deletes, but the journal owns it: it survives, closed, every pass.
689func TestRollbackRecordDeletedFenceDeferredRowSurvivesRemoval(t *testing.T) {
690 x := abortedImport(t)
691 if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil {
692 t.Fatal(err)
693 }
694 x.box.proceed = false
695 for i := 0; i < 2; i++ {
696 x.pass(t)
697 got := mustGet(t, x.reg, x.rec.Host)
698 if got.Kind != "instance" || got.ImportGate != registry.ImportGateClosed || !got.Provisioned {
699 t.Fatalf("pass %d: row = %+v, want the closed import row", i, got)
700 }
701 if st := x.journalState(t); st != registry.JournalRollingBack {
702 t.Fatalf("pass %d: journal = %s", i, st)
703 }
704 }
705 if !x.box.local || !x.box.replica || count(x.box.steps, "remove-local") != 0 {
706 t.Fatalf("a deferred rollback removed or deleted something: %q", x.box.steps)
707 }
708 x.box.proceed = true
709 x.pass(t)
710 if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) {
711 t.Fatalf("row = %+v", got)
712 }
713 if x.journalState(t) != "" || x.box.local || x.box.replica {
714 t.Fatalf("not finished: journal %q local %v replica %v", x.journalState(t), x.box.local, x.box.replica)
715 }
716}
717
718// The rollback stops the instance before it removes or deletes anything,
719// restores the route before the bucket is touched, and the removal phase,
720// which runs on the route list read before the rollback, leaves the
721// restored legacy row alone (the journaled-host skip): here the record is
722// gone, so without the skip the stale Provisioned import row would have
723// its host removed, taking the legacy row with it.
724func TestRollbackOrderAndTheRemovalPhaseKeepsTheRestoredRow(t *testing.T) {
725 x := abortedImport(t)
726 if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil {
727 t.Fatal(err)
728 }
729 x.pass(t)
730 want := []string{"gen", "fence", "remove-local", "lift", "delete-snapshot", "delete-replica"}
731 if !reflect.DeepEqual(x.box.steps, want) {
732 t.Fatalf("steps = %q, want %q", x.box.steps, want)
733 }
734 if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) {
735 t.Fatalf("row after the pass = %+v, want the original", got)
736 }
737 if x.journalState(t) != "" {
738 t.Fatalf("journal = %s", x.journalState(t))
739 }
740}
741
742// insertJournal writes a journal row straight into the registry file, as
743// a cold restore of an older snapshot would bring it back.
744func insertJournal(t *testing.T, path string, j registry.ImportJournal) {
745 t.Helper()
746 orig, err := json.Marshal(j.Original)
747 if err != nil {
748 t.Fatal(err)
749 }
750 db, err := sql.Open("sqlite", path)
751 if err != nil {
752 t.Fatal(err)
753 }
754 defer db.Close()
755 if _, err := db.Exec(`INSERT INTO import_journal (host, seed_id, original, record_key, state, fence_gen, at) VALUES (?,?,?,?,?,?,?)`,
756 j.Host, j.SeedID, string(orig), j.RecordKey, j.State, j.FenceGen, j.At); err != nil {
757 t.Fatal(err)
758 }
759}
760
761// The Task 4 review's stale journal row: the import was rolled back and
762// its record finished, the host was created again, and then a journal row
763// for the old import comes back (a cold restore of an older snapshot).
764// Whatever state the row comes back in, nothing the new instance owns is
765// touched: not its row, its record, its process, its local data or its
766// replica.
767func TestRollbackOfAStaleJournalTouchesNothingTheNewInstanceOwns(t *testing.T) {
768 for _, owner := range []string{"the same app", "another app"} {
769 for _, state := range []string{registry.JournalSwapped, registry.JournalRollingBack, registry.JournalCleanupPending} {
770 t.Run(owner+"/"+state, func(t *testing.T) {
771 x := abortedImport(t)
772 stale, _ := x.reg.ImportJournal(x.rec.Host)
773 x.pass(t)
774 x.wantRolledBack(t)
775 // The console finishes A and the host is created again.
776 if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil {
777 t.Fatal(err)
778 }
779 next := shopRecord(x.rec.Account)
780 if owner == "another app" {
781 next.App = "other"
782 if err := PutManifest(ctx, x.raw, next.Account, "other", Manifest{Schema: 1, App: "other", Domain: "other.example.com"}); err != nil {
783 t.Fatal(err)
784 }
785 }
786 if err := CreateRecord(ctx, x.raw, next); err != nil {
787 t.Fatal(err)
788 }
789 // The new instance's row, as the box serves it.
790 if err := x.reg.Delete(next.Host); err != nil {
791 t.Fatal(err)
792 }
793 x.pass(t)
794 newRow := mustGet(t, x.reg, next.Host)
795 if newRow.App != next.App || !newRow.Provisioned || newRow.ImportGate != "" {
796 t.Fatalf("the new instance was not minted: %+v", newRow)
797 }
798 x.box.local, x.box.replica = true, true // its own data now
799 x.box.original = *newRow // what any bucket delete must find at the host
800 newRec, _ := GetRecord(ctx, x.raw, next.Account, next.App, next.Host)
801
802 j := *stale
803 j.State = state
804 if state != registry.JournalSwapped {
805 j.FenceGen = 7
806 }
807 insertJournal(t, x.path, j)
808 x.box.steps = nil
809 logged := captureLog(t)
810 x.pass(t)
811 x.pass(t)
812
813 if got := mustGet(t, x.reg, next.Host); !reflect.DeepEqual(*got, *newRow) {
814 t.Fatalf("the new instance's row changed:\n got %+v\nwant %+v", got, newRow)
815 }
816 if got, _ := GetRecord(ctx, x.raw, next.Account, next.App, next.Host); !reflect.DeepEqual(got, newRec) {
817 t.Fatalf("the new instance's record changed: %+v", got)
818 }
819 for _, s := range x.box.steps {
820 // A lift names only the stale row's own generation,
821 // which no fence on the host holds.
822 if s != "delete-snapshot" && s != "lift" {
823 t.Fatalf("the stale journal reached the box: %q\n%s", x.box.steps, logged)
824 }
825 }
826 if !x.box.local || !x.box.replica {
827 t.Fatalf("the new instance's data went: local %v replica %v", x.box.local, x.box.replica)
828 }
829 // A stale row that would act on the route is held; one at
830 // cleanup-pending finishes its own seed's cleanup and goes.
831 if st := x.journalState(t); (state == registry.JournalCleanupPending) != (st == "") {
832 t.Fatalf("journal = %q after a stale %s row", st, state)
833 }
834 why := "the route at the host is not the import's"
835 if state == registry.JournalCleanupPending {
836 why = "the host's replica is left alone"
837 }
838 if !strings.Contains(logged.String(), why) {
839 t.Fatalf("not held for the reason %q:\n%s", why, logged)
840 }
841 })
842 }
843 }
844}
845
846// The rollback tolerates a missing import row (the Task 4 review's carry):
847// something removed it, so no route is put back, and the cleanup carries
848// on. Its local data is still removed under the fence (fix rounds 1 and 2):
849// the database's marker names this seed and the host has no row.
850func TestRollbackOfAMissingRowSkipsTheRestore(t *testing.T) {
851 x := abortedImport(t)
852 if err := x.reg.Delete(x.rec.Host); err != nil {
853 t.Fatal(err)
854 }
855 x.box.noRoute = true
856 x.pass(t)
857 if rt, _ := x.reg.Get(x.rec.Host); rt != nil {
858 t.Fatalf("a route came back: %+v", rt)
859 }
860 want := []string{"gen", "fence", "read-marker", "remove-local", "lift", "delete-snapshot", "delete-replica"}
861 if !reflect.DeepEqual(x.box.steps, want) {
862 t.Fatalf("steps = %q, want %q", x.box.steps, want)
863 }
864 if x.journalState(t) != "" || x.box.replica || x.box.local {
865 t.Fatalf("journal %q replica %v local %v", x.journalState(t), x.box.replica, x.box.local)
866 }
867 if st := seedState(t, x.raw, x.rec); st.State != seeds.StateRolledBack {
868 t.Fatalf("record = %s", st.State)
869 }
870}
871
872// Step 6 deletes the host's replica only when it can tell the replica is
873// this import's: a record it cannot read, a host listing it cannot finish,
874// or a record naming this seed in a state that is not abandoned all defer
875// the cleanup; the journal stays at cleanup-pending.
876func TestRollbackDefersTheReplicaDeleteWhenItCannotTell(t *testing.T) {
877 for _, c := range []string{"the record read fails", "the record names this seed, placed"} {
878 t.Run(c, func(t *testing.T) {
879 x := abortedImport(t)
880 if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 9); err != nil {
881 t.Fatal(err)
882 }
883 if err := x.reg.RestoreImportRoute(x.rec.Host, x.rec.Seed.ID); err != nil {
884 t.Fatal(err)
885 }
886 key := RecordKey(x.rec.Account, x.rec.App, x.rec.Host)
887 switch c {
888 case "the record read fails":
889 // Only the step-6 re-read fails: the pass's own read works.
890 x.store = &failNthGet{Store: x.store, key: key, n: 2}
891 default:
892 placed := x.rec
893 seed := *x.rec.Seed
894 seed.State = seeds.StatePlaced
895 placed.Seed = &seed
896 putReplace(t, x.raw, key, mustJSON(t, placed))
897 }
898 logged := captureLog(t)
899 x.pass(t)
900 if st := x.journalState(t); st != registry.JournalCleanupPending {
901 t.Fatalf("journal = %q", st)
902 }
903 if !strings.Contains(logged.String(), "cleanup deferred") {
904 t.Fatalf("not deferred:\n%s", logged)
905 }
906 if !x.box.replica || count(x.box.steps, "delete-replica") != 0 {
907 t.Fatalf("the replica was deleted: %q", x.box.steps)
908 }
909 })
910 }
911}
912
913// failNthGet fails every read of one key from the nth on.
914type failNthGet struct {
915 release.Store
916 key string
917 n int
918 seen int
919}
920
921func (f *failNthGet) Get(ctx context.Context, key string) ([]byte, string, error) {
922 if key == f.key {
923 f.seen++
924 if f.seen >= f.n {
925 return nil, "", errors.New("connection reset")
926 }
927 }
928 return f.Store.Get(ctx, key)
929}
930
931// Without every box hook the rollback does nothing at all: the import
932// stays journaled, gated and untouched.
933func TestRollbackNeedsEveryHook(t *testing.T) {
934 x := abortedImport(t)
935 x.cfg.Seeds.DeleteReplica = nil
936 row := *mustGet(t, x.reg, x.rec.Host)
937 x.pass(t)
938 if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, row) {
939 t.Fatalf("row changed: %+v", got)
940 }
941 if st := x.journalState(t); st != registry.JournalSwapped || len(x.box.steps) != 0 {
942 t.Fatalf("journal %q steps %q", st, x.box.steps)
943 }
944}
945
946// The route is read again under the fence: a row that stopped being the
947// import's while the host was being fenced (removed and replaced by
948// another, here) is not removed, the fence is lifted, and the rollback
949// holds with the journal at rolling-back.
950func TestRollbackRereadsTheRouteUnderTheFence(t *testing.T) {
951 x := abortedImport(t)
952 other := addrRow(x.rec, x.rec.Account)
953 other.Addr = "127.0.0.1:9999"
954 fence := x.cfg.Seeds.FenceForRollback
955 x.cfg.Seeds.FenceForRollback = func(ctx context.Context, host string, gen uint64) (bool, error) {
956 proceed, err := fence(ctx, host, gen)
957 if err := x.reg.Delete(host); err != nil {
958 t.Fatal(err)
959 }
960 if err := x.reg.Upsert(other); err != nil {
961 t.Fatal(err)
962 }
963 return proceed, err
964 }
965 x.pass(t)
966 placed := mustGet(t, x.reg, x.rec.Host)
967 if placed.Addr != other.Addr || placed.ImportGate != "" {
968 t.Fatalf("the replacing row changed: %+v", placed)
969 }
970 if count(x.box.steps, "remove-local") != 0 || !x.box.local || len(x.box.fenced) != 0 {
971 t.Fatalf("steps %q local %v fenced %v", x.box.steps, x.box.local, x.box.fenced)
972 }
973 if st := x.journalState(t); st != registry.JournalRollingBack {
974 t.Fatalf("journal = %q", st)
975 }
976}
977
978// Preflight M3: a journal row whose seed id fails seeds.Check names no
979// snapshot and no replica of this import: the cleanup deletes nothing in
980// the bucket and only retires the journal row.
981func TestRollbackCleanupOfAnInvalidSeedIDDeletesNothing(t *testing.T) {
982 s, sqid := shopSetup(t)
983 rec := importRecord(t, s, sqid)
984 reg, path := regAt(t, addrRow(rec, sqid))
985 original := *mustGet(t, reg, rec.Host)
986 insertJournal(t, path, registry.ImportJournal{Host: rec.Host, SeedID: "../not-a-seed", Original: original,
987 RecordKey: RecordKey(rec.Account, rec.App, rec.Host), State: registry.JournalCleanupPending, FenceGen: 3, At: 1})
988 cfg := importCfg(t, nil)
989 b := newRollbackBox(t, reg, original)
990 b.hooks(cfg.Seeds)
991 cfg.Seeds.Data = faultData{cfg.Seeds.Data, b}
992 if err := Reconcile(ctx, s, reg, cfg); err != nil {
993 t.Fatal(err)
994 }
995 if len(b.steps) != 0 || !b.replica {
996 t.Fatalf("steps %q replica %v", b.steps, b.replica)
997 }
998 if j, _ := reg.ImportJournal(rec.Host); j != nil {
999 t.Fatalf("journal = %+v", j)
1000 }
1001 if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, original) {
1002 t.Fatalf("row = %+v", got)
1003 }
1004}
1005
1006// Fix round 2: with the import's row gone, its local database is decided
1007// from the file. Removed when its marker names this seed (whoever holds
1008// the host's record, here another app's); left when the marker names
1009// another seed or there is none; and a marker that cannot be read lifts
1010// the fence and defers. The replica still follows step 6's record test:
1011// another record holds the host, so it is kept.
1012func TestRollbackOfAMissingRowDecidesTheLocalDataFromItsMarker(t *testing.T) {
1013 otherSeed, _ := seeds.NewID()
1014 for _, c := range []struct {
1015 name string
1016 marker func(x *rollbackScene) string
1017 markErr error
1018 removed bool
1019 defers bool
1020 logged string
1021 }{
1022 {name: "the marker names this seed", marker: func(x *rollbackScene) string { return x.rec.Seed.ID }, removed: true,
1023 logged: "its local data is removed"},
1024 {name: "the marker names another seed", marker: func(*rollbackScene) string { return otherSeed },
1025 logged: "its seed marker names \"" + otherSeed + "\", not this seed"},
1026 {name: "no marker", marker: func(*rollbackScene) string { return "" },
1027 logged: "its seed marker names \"\", not this seed"},
1028 {name: "the marker cannot be read", markErr: errors.New("permission denied"), defers: true,
1029 logged: "the seed marker beside"},
1030 } {
1031 t.Run(c.name, func(t *testing.T) {
1032 x := abortedImport(t)
1033 if err := x.reg.Delete(x.rec.Host); err != nil {
1034 t.Fatal(err)
1035 }
1036 x.box.noRoute = true
1037 if c.marker != nil {
1038 x.box.marker = c.marker(x)
1039 }
1040 x.box.markErr = c.markErr
1041 other := shopRecord(x.rec.Account)
1042 other.App = "other"
1043 if err := CreateRecord(ctx, x.raw, other); err != nil {
1044 t.Fatal(err)
1045 }
1046 logged := captureLog(t)
1047 x.pass(t)
1048 if !strings.Contains(logged.String(), c.logged) {
1049 t.Fatalf("not logged %q:\n%s", c.logged, logged)
1050 }
1051 if x.box.local == c.removed || (count(x.box.steps, "remove-local") == 1) != c.removed {
1052 t.Fatalf("local %v steps %q; want removed %v", x.box.local, x.box.steps, c.removed)
1053 }
1054 // The instance key follows the database, by the same marker:
1055 // retired with it, left with it (merge review S1).
1056 if x.box.key == c.removed || (len(x.box.retired) == 1) != c.removed {
1057 t.Fatalf("key left %v, retire calls %q; want retired %v", x.box.key, x.box.retired, c.removed)
1058 }
1059 if len(x.box.fenced) != 0 {
1060 t.Fatalf("a fence was left: %v", x.box.fenced)
1061 }
1062 if c.defers {
1063 if st := x.journalState(t); st != registry.JournalRollingBack {
1064 t.Fatalf("journal = %q, want the rollback waiting at rolling-back", st)
1065 }
1066 if want := []string{"gen", "fence", "read-marker", "lift"}; !reflect.DeepEqual(x.box.steps, want) {
1067 t.Fatalf("steps = %q, want %q", x.box.steps, want)
1068 }
1069 return
1070 }
1071 if x.journalState(t) != "" || !x.box.replica {
1072 t.Fatalf("journal %q replica %v: the cleanup did not finish, or deleted another record's replica",
1073 x.journalState(t), x.box.replica)
1074 }
1075 })
1076 }
1077}
1078
1079// Review Minor 1: a fence hook that fails is followed by a lift of this
1080// rollback's generation, in case it fenced before it failed.
1081func TestRollbackLiftsTheFenceWhenTheFenceHookFails(t *testing.T) {
1082 x := abortedImport(t)
1083 fence := x.cfg.Seeds.FenceForRollback
1084 x.cfg.Seeds.FenceForRollback = func(ctx context.Context, host string, gen uint64) (bool, error) {
1085 _, _ = fence(ctx, host, gen) // installs it
1086 return true, errors.New("stop evidence unreadable")
1087 }
1088 x.pass(t)
1089 if len(x.box.fenced) != 0 || count(x.box.steps, "remove-local") != 0 {
1090 t.Fatalf("fenced %v steps %q", x.box.fenced, x.box.steps)
1091 }
1092}
1093
1094// Fix round 1, I1, the provision half of the end-to-end check: with the
1095// import's row gone (and its marker naming this seed), the local removal
1096// is aimed at the host's default
1097// nested database path, which is exactly where a plain instance created at
1098// the host afterwards is minted. (The activator half, seedpurge_linux_test.go,
1099// shows a plain wake at that path finds no file once PurgeSeedLocal ran.)
1100func TestRollbackOfAMissingRowPurgesWhereTheNextInstanceIsMinted(t *testing.T) {
1101 s, sqid := shopSetup(t)
1102 rec := importRecord(t, s, sqid)
1103 reg := regWith(t, addrRow(rec, sqid))
1104 cfg := importCfg(t, nil)
1105 rr := &rollbackRecorder{proceed: true}
1106 rr.hooks(cfg.Seeds)
1107 rr.marker = rec.Seed.ID
1108 var purged []string
1109 cfg.Seeds.RemoveLocal = func(rt registry.Route) error { purged = append(purged, rt.DBPath); return nil }
1110 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1111 t.Fatal(err)
1112 }
1113 if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
1114 t.Fatal(err)
1115 }
1116 if err := reg.Delete(rec.Host); err != nil {
1117 t.Fatal(err)
1118 }
1119 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1120 t.Fatal(err)
1121 }
1122 if want := []string{NestedDBPath(cfg.DataDir, rec.Host)}; !reflect.DeepEqual(purged, want) {
1123 t.Fatalf("purged %q, want %q", purged, want)
1124 }
1125 if err := DeleteRecord(ctx, s, rec.Account, rec.App, rec.Host); err != nil {
1126 t.Fatal(err)
1127 }
1128 if err := CreateRecord(ctx, s, shopRecord(sqid)); err != nil {
1129 t.Fatal(err)
1130 }
1131 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1132 t.Fatal(err)
1133 }
1134 if next := mustGet(t, reg, rec.Host); next.DBPath != purged[0] || next.SeedSource != "" {
1135 t.Fatalf("the plain instance's row = %+v, want it at %s", next, purged[0])
1136 }
1137}
1138
1139// Review Minor 1: a fence this rollback kept after a failed step is lifted
1140// when, by the next pass, the row at the host is no longer the import's.
1141func TestRollbackLiftsAKeptFenceWhenTheRowIsNoLongerTheImports(t *testing.T) {
1142 x := abortedImport(t)
1143 x.box.fault = "remove-local" // fails once; the fence is kept for the resume
1144 x.pass(t)
1145 if len(x.box.fenced) != 1 {
1146 t.Fatalf("fenced %v, want this rollback's fence kept", x.box.fenced)
1147 }
1148 other := addrRow(x.rec, x.rec.Account)
1149 other.Addr = "127.0.0.1:9999"
1150 if err := x.reg.Delete(x.rec.Host); err != nil {
1151 t.Fatal(err)
1152 }
1153 if err := x.reg.Upsert(other); err != nil {
1154 t.Fatal(err)
1155 }
1156 x.pass(t)
1157 if len(x.box.fenced) != 0 || x.journalState(t) != registry.JournalRollingBack {
1158 t.Fatalf("fenced %v journal %q", x.box.fenced, x.journalState(t))
1159 }
1160}
1161
1162// Fix round 2: a row that appears at the host while the marker is read
1163// (not expected while journaled) keeps the local database, lifts the
1164// fence, and leaves the rollback waiting for the next pass to judge it.
1165func TestRollbackOfAMissingRowLeavesTheDataWhenARowAppears(t *testing.T) {
1166 x := abortedImport(t)
1167 if err := x.reg.Delete(x.rec.Host); err != nil {
1168 t.Fatal(err)
1169 }
1170 x.box.noRoute = true
1171 read := x.cfg.Seeds.LocalSeedID
1172 x.cfg.Seeds.LocalSeedID = func(dbPath string) (string, error) {
1173 if err := x.reg.Upsert(addrRow(x.rec, x.rec.Account)); err != nil {
1174 t.Fatal(err)
1175 }
1176 return read(dbPath)
1177 }
1178 x.pass(t)
1179 if !x.box.local || count(x.box.steps, "remove-local") != 0 || len(x.box.fenced) != 0 {
1180 t.Fatalf("local %v steps %q fenced %v", x.box.local, x.box.steps, x.box.fenced)
1181 }
1182 if st := x.journalState(t); st != registry.JournalRollingBack {
1183 t.Fatalf("journal = %q", st)
1184 }
1185}
1186
1187// A generation that cannot be minted (the box's source failed, or answered
1188// 0, which is no fence's) closes nothing: the journal stays swapped with
1189// no generation stored, so a later pass mints a real one. A 0 stored there
1190// would read as "ours already" on any unfenced host.
1191func TestRollbackWaitsForAFenceGeneration(t *testing.T) {
1192 for name, mint := range map[string]func() (uint64, error){
1193 "an error": func() (uint64, error) { return 0, errors.New("registry busy") },
1194 "zero": func() (uint64, error) { return 0, nil },
1195 } {
1196 t.Run(name, func(t *testing.T) {
1197 x := abortedImport(t)
1198 x.cfg.Seeds.NextFenceGen = mint
1199 x.pass(t)
1200 j, _ := x.reg.ImportJournal(x.rec.Host)
1201 if j == nil || j.State != registry.JournalSwapped || j.FenceGen != 0 {
1202 t.Fatalf("journal = %+v, want swapped with no generation", j)
1203 }
1204 if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate == registry.ImportGateClosed {
1205 t.Fatalf("the gate closed without a generation: %+v", got)
1206 }
1207 if len(x.box.steps) != 0 {
1208 t.Fatalf("steps = %q", x.box.steps)
1209 }
1210 })
1211 }
1212}
1213
1214// Merge review S1: the swap records what the host's own env file held
1215// before the import, in the journal, and swaps only when it could read it.
1216func TestSwapRecordsTheEnvFileBeforeTheImport(t *testing.T) {
1217 for _, before := range []string{registry.EnvBeforeAbsent, registry.EnvBeforeNoKey, registry.EnvBeforeKey} {
1218 t.Run(before, func(t *testing.T) {
1219 s, sqid := shopSetup(t)
1220 rec := importRecord(t, s, sqid)
1221 reg := regWith(t, addrRow(rec, sqid))
1222 cfg := importCfg(t, nil)
1223 var asked []string
1224 cfg.Seeds.EnvBefore = func(db, host string) (string, error) {
1225 // Asked before the import's row exists.
1226 if rt := mustGet(t, reg, rec.Host); rt.ImportGate != "" || rt.Kind != "service" && rt.Addr == "" {
1227 t.Errorf("the env file was read after the swap: %+v", rt)
1228 }
1229 asked = append(asked, db+" "+host)
1230 return before, nil
1231 }
1232 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1233 t.Fatal(err)
1234 }
1235 j, err := reg.ImportJournal(rec.Host)
1236 if err != nil || j == nil || j.EnvBefore != before {
1237 t.Fatalf("journal = %+v, %v; want env before %q", j, err, before)
1238 }
1239 if want := []string{NestedDBPath("/data", rec.Host) + " " + rec.Host}; !reflect.DeepEqual(asked, want) {
1240 t.Fatalf("asked %q, want %q", asked, want)
1241 }
1242 })
1243 }
1244 for name, hook := range map[string]func(string, string) (string, error){
1245 "the file cannot be read": func(string, string) (string, error) { return "", errors.New("a symlink") },
1246 "no reader": nil,
1247 } {
1248 t.Run(name, func(t *testing.T) {
1249 s, sqid := shopSetup(t)
1250 rec := importRecord(t, s, sqid)
1251 reg := regWith(t, addrRow(rec, sqid))
1252 original := *mustGet(t, reg, rec.Host)
1253 cfg := importCfg(t, nil)
1254 cfg.Seeds.EnvBefore = hook
1255 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1256 t.Fatal(err)
1257 }
1258 if j, _ := reg.ImportJournal(rec.Host); j != nil {
1259 t.Fatalf("swapped without knowing what the env file held: %+v", j)
1260 }
1261 if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, original) {
1262 t.Fatalf("row = %+v, want the original untouched", got)
1263 }
1264 if st := seedState(t, s, rec); st.State != seeds.StateReady {
1265 t.Fatalf("record = %s, want ready: a deferred swap refuses nothing", st.State)
1266 }
1267 })
1268 }
1269}
1270
1271// Merge review S1: the rollback hands the journal's record of the env file
1272// to the key's retirement, for the import's own row, whatever it says. A
1273// key that was there before the import is the hook's to leave (the
1274// activator's RetireSeedKey; its own tests cover the file).
1275func TestRollbackRetiresTheKeyByWhatTheSwapRecorded(t *testing.T) {
1276 for _, before := range []string{registry.EnvBeforeAbsent, registry.EnvBeforeNoKey, registry.EnvBeforeKey} {
1277 t.Run(before, func(t *testing.T) {
1278 s, sqid := shopSetup(t)
1279 rec := importRecord(t, s, sqid)
1280 reg := regWith(t, addrRow(rec, sqid))
1281 cfg := importCfg(t, nil)
1282 cfg.Seeds.EnvBefore = func(string, string) (string, error) { return before, nil }
1283 rr := &rollbackRecorder{proceed: true}
1284 rr.hooks(cfg.Seeds)
1285 Reconcile(ctx, s, reg, cfg) // swap
1286 if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil {
1287 t.Fatal(err)
1288 }
1289 if err := Reconcile(ctx, s, reg, cfg); err != nil {
1290 t.Fatal(err)
1291 }
1292 want := []string{NestedDBPath("/data", rec.Host) + " " + rec.Host + " " + before}
1293 if !reflect.DeepEqual(rr.retired, want) {
1294 t.Fatalf("retired %q, want %q", rr.retired, want)
1295 }
1296 })
1297 }
1298}
1299
1300// A retirement that fails keeps the rollback where it was, fenced, with
1301// the local data and the gated row in place; the next pass tries again.
1302func TestRollbackWaitsWhileTheKeyCannotBeRetired(t *testing.T) {
1303 x := abortedImport(t)
1304 retire := x.cfg.Seeds.RetireKey
1305 fail := true
1306 x.cfg.Seeds.RetireKey = func(db, host, before string) error {
1307 if fail {
1308 return errors.New("the env file is a symlink")
1309 }
1310 return retire(db, host, before)
1311 }
1312 x.pass(t)
1313 if st := x.journalState(t); st != registry.JournalRollingBack {
1314 t.Fatalf("journal = %q, want rolling-back", st)
1315 }
1316 if !x.box.local || !x.box.key || count(x.box.steps, "remove-local") != 0 {
1317 t.Fatalf("local %v key %v steps %q: the rollback went on past a key it could not retire", x.box.local, x.box.key, x.box.steps)
1318 }
1319 if got := mustGet(t, x.reg, x.rec.Host); got.ImportGate != registry.ImportGateClosed {
1320 t.Fatalf("row = %+v, want the closed import row", got)
1321 }
1322 fail = false
1323 x.pass(t)
1324 x.wantRolledBack(t)
1325}
1326