| new file mode 100644 |
| index 000000000..5cc2ac9b3 |
| --- /dev/null |
| +++ b/internal/provision/seedrollback_test.go |
| @@ -0,0 +1,931 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "database/sql" |
| + "encoding/json" |
| + "errors" |
| + "os" |
| + "reflect" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/seeds" |
| +) |
| + |
| +// rollbackRecorder is the brief's fake: it records the box effects in the |
| +// order they happen. |
| +type rollbackRecorder struct { |
| + steps []string |
| + proceed bool |
| + fenceErr error |
| + gen uint64 |
| +} |
| + |
| +func (r *rollbackRecorder) hooks(h *SeedHooks) { |
| + h.NextFenceGen = func() uint64 { r.gen++; return r.gen } |
| + h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) { |
| + r.steps = append(r.steps, "fence") |
| + return r.proceed, r.fenceErr |
| + } |
| + h.LiftFence = func(host string, gen uint64) { r.steps = append(r.steps, "lift") } |
| + h.RemoveLocal = func(rt registry.Route) error { r.steps = append(r.steps, "remove-local:"+rt.Kind); return nil } |
| + h.DeleteReplica = func(_ context.Context, host string) error { r.steps = append(r.steps, "delete-replica"); return nil } |
| +} |
| + |
| +func TestRollbackAfterAbortRestoresTheRowExactly(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + original, _ := reg.Get(rec.Host) |
| + cfg := importCfg(t, nil) |
| + rr := &rollbackRecorder{proceed: true} |
| + rr.hooks(cfg.Seeds) |
| + cfg.Seeds.Data.Put(ctx, seeds.SnapshotKey(rec.Host, rec.Seed.ID), []byte("db"), "") |
| + other, _ := seeds.NewID() |
| + cfg.Seeds.Data.Put(ctx, seeds.SnapshotKey(rec.Host, other), []byte("db"), "") |
| + Reconcile(ctx, s, reg, cfg) // swap |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) { |
| + t.Fatalf("restored %+v, want %+v", got, original) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| + want := []string{"fence", "remove-local:instance", "lift", "delete-replica"} |
| + if !reflect.DeepEqual(rr.steps, want) { |
| + t.Fatalf("steps = %q, want %q", rr.steps, want) |
| + } |
| + if _, _, err := cfg.Seeds.Data.Get(ctx, seeds.SnapshotKey(rec.Host, rec.Seed.ID)); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatal("this seed's snapshot survived") |
| + } |
| + if _, _, err := cfg.Seeds.Data.Get(ctx, seeds.SnapshotKey(rec.Host, other)); err != nil { |
| + t.Fatal("another seed's snapshot was deleted") |
| + } |
| + r, _ := GetRecord(ctx, s, rec.Account, rec.App, rec.Host) |
| + if r.Seed.State != seeds.StateRolledBack { |
| + t.Fatalf("record = %s", r.Seed.State) |
| + } |
| +} |
| + |
| +func TestRollbackForAGoneOrReplacedRecord(t *testing.T) { |
| + for _, mutate := range []string{"deleted", "another seed"} { |
| + t.Run(mutate, func(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + original, _ := reg.Get(rec.Host) |
| + cfg := importCfg(t, nil) |
| + rr := &rollbackRecorder{proceed: true} |
| + rr.hooks(cfg.Seeds) |
| + Reconcile(ctx, s, reg, cfg) |
| + if err := DeleteRecord(ctx, s, rec.Account, rec.App, rec.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + var next Record |
| + if mutate == "another seed" { |
| + next = importRecord(t, s, sqid) // the same host, a new seed id, state ready |
| + } |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, *original) { |
| + t.Fatalf("restored %+v", got) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| + // A record naming another seed holds the host: the host's |
| + // replica is not this rollback's to delete, and that record |
| + // keeps its own state. |
| + wantSteps := []string{"fence", "remove-local:instance", "lift", "delete-replica"} |
| + if mutate == "another seed" { |
| + wantSteps = wantSteps[:3] |
| + if st := seedState(t, s, next); st.ID != next.Seed.ID || st.State != seeds.StateReady { |
| + t.Fatalf("the new record's seed = %+v", st) |
| + } |
| + } |
| + if !reflect.DeepEqual(rr.steps, wantSteps) { |
| + t.Fatalf("steps = %q, want %q", rr.steps, wantSteps) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +func TestRollbackDefersWhenTheFenceIsNotOurs(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + cfg := importCfg(t, nil) |
| + rr := &rollbackRecorder{proceed: false} |
| + rr.hooks(cfg.Seeds) |
| + Reconcile(ctx, s, reg, cfg) |
| + SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)) |
| + Reconcile(ctx, s, reg, cfg) |
| + j, _ := reg.ImportJournal(rec.Host) |
| + got := mustGet(t, reg, rec.Host) |
| + if j == nil || j.State != registry.JournalRollingBack || got.ImportGate != registry.ImportGateClosed || got.Kind != "instance" { |
| + t.Fatalf("after a deferred fence: journal %+v row %+v; want rolling-back, gate closed, row kept", j, got) |
| + } |
| + if !reflect.DeepEqual(rr.steps, []string{"fence"}) { |
| + t.Fatalf("steps = %q: a deferred fence went on to remove or delete", rr.steps) |
| + } |
| + // Next pass the fence proceeds, with the SAME generation. |
| + gen := j.FenceGen |
| + rr.proceed = true |
| + Reconcile(ctx, s, reg, cfg) |
| + if rr.gen != gen { |
| + t.Fatalf("a resumed rollback minted a new generation: %d, journal had %d", rr.gen, gen) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| +} |
| + |
| +// A fence hook that answers with an error defers, whatever proceed says. |
| +func TestRollbackDefersOnAFenceError(t *testing.T) { |
| + x := abortedImport(t) |
| + rr := &rollbackRecorder{proceed: true, fenceErr: errors.New("stop unproven")} |
| + rr.hooks(x.cfg.Seeds) |
| + x.pass(t) |
| + if j, _ := x.reg.ImportJournal(x.rec.Host); j == nil || j.State != registry.JournalRollingBack { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| + if !reflect.DeepEqual(rr.steps, []string{"fence"}) { |
| + t.Fatalf("steps = %q", rr.steps) |
| + } |
| +} |
| + |
| +func TestCleanupPendingNeverFences(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + cfg := importCfg(t, nil) |
| + rr := &rollbackRecorder{proceed: true} |
| + rr.hooks(cfg.Seeds) |
| + Reconcile(ctx, s, reg, cfg) |
| + SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)) |
| + // Crash after the route restore, before the bucket cleanup. |
| + if err := reg.CloseImportGate(rec.Host, rec.Seed.ID, 9); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := reg.RestoreImportRoute(rec.Host, rec.Seed.ID); err != nil { |
| + t.Fatal(err) |
| + } |
| + rr.steps = nil |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !reflect.DeepEqual(rr.steps, []string{"delete-replica"}) { |
| + t.Fatalf("steps = %q; a cleanup-pending resume fenced, stopped or removed something", rr.steps) |
| + } |
| + if rr.gen != 0 { |
| + t.Fatalf("a cleanup-pending resume minted a fence generation") |
| + } |
| +} |
| + |
| +func TestJournalDoesNotRollBackOnAReadFailure(t *testing.T) { |
| + for _, how := range []string{"unparseable body", "read error"} { |
| + t.Run(how, func(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg := regWith(t, addrRow(rec, sqid)) |
| + cfg := importCfg(t, nil) |
| + rr := &rollbackRecorder{proceed: true} |
| + rr.hooks(cfg.Seeds) |
| + Reconcile(ctx, s, reg, cfg) |
| + var store release.Store = s |
| + key := RecordKey(rec.Account, rec.App, rec.Host) |
| + if how == "unparseable body" { |
| + // The record body becomes unreadable: protected, never "gone". |
| + putReplace(t, s, key, []byte("{not json")) |
| + } else { |
| + // preflight m7: a Get that fails (readFailed), not a bad body. |
| + store = getFailsFor{s, key} |
| + } |
| + if err := Reconcile(ctx, store, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(rr.steps) != 0 || rr.gen != 0 { |
| + t.Fatalf("an unreadable record started a rollback: steps %q", rr.steps) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j == nil || j.State != registry.JournalSwapped { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| + if got := mustGet(t, reg, rec.Host); got.Kind != "instance" { |
| + t.Fatal("the removal phase touched a journaled host") |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// getFailsFor fails every read of one key with something other than "not |
| +// found". |
| +type getFailsFor struct { |
| + release.Store |
| + key string |
| +} |
| + |
| +func (g getFailsFor) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == g.key { |
| + return nil, "", errors.New("connection reset") |
| + } |
| + return g.Store.Get(ctx, key) |
| +} |
| + |
| +// ---- the box side, with the step order checked as it happens ------------- |
| + |
| +// crashed is the panic a fault raises to kill the "process" mid-pass. |
| +type crashed struct{} |
| + |
| +// rollbackBox is the box's side of the rollback: an in-memory fence (lost |
| +// when the process dies), the import's local data, and the host's replica. |
| +// Every hook checks, when it is called, the order the spec requires: the |
| +// fence comes after the gate is closed and under the journal's own |
| +// generation; nothing local is removed unless the host is fenced and the |
| +// journal is still rolling back; nothing in the bucket is deleted before |
| +// the original route is back. |
| +type rollbackBox struct { |
| + t *testing.T |
| + reg *registry.Registry |
| + original registry.Route |
| + steps []string |
| + proceed bool |
| + gen uint64 |
| + fenced map[string]uint64 |
| + local bool |
| + replica bool |
| + fault string // the step that fails, once |
| + crash bool // the fault kills the process (a panic) rather than returning an error |
| + noRoute bool // the import's row was removed: no route comes back |
| +} |
| + |
| +func newRollbackBox(t *testing.T, reg *registry.Registry, original registry.Route) *rollbackBox { |
| + return &rollbackBox{t: t, reg: reg, original: original, proceed: true, fenced: map[string]uint64{}, local: true, replica: true} |
| +} |
| + |
| +func (b *rollbackBox) trip(step string) error { |
| + if b.fault != step { |
| + return nil |
| + } |
| + b.fault = "" |
| + if b.crash { |
| + panic(crashed{}) |
| + } |
| + return errors.New("injected failure at " + step) |
| +} |
| + |
| +// restart is the process dying: the activator's fences are in memory. |
| +func (b *rollbackBox) restart() { b.fenced = map[string]uint64{} } |
| + |
| +func (b *rollbackBox) journal() *registry.ImportJournal { |
| + j, err := b.reg.ImportJournal(b.original.Host) |
| + if err != nil { |
| + b.t.Fatal(err) |
| + } |
| + return j |
| +} |
| + |
| +func (b *rollbackBox) row() *registry.Route { |
| + rt, err := b.reg.Get(b.original.Host) |
| + if err != nil { |
| + b.t.Fatal(err) |
| + } |
| + return rt |
| +} |
| + |
| +// wantRestored: nothing is deleted from the bucket before the original |
| +// route is back and the journal says cleanup-pending (or is gone, for the |
| +// no-journal finish). |
| +func (b *rollbackBox) wantRestored(what string) { |
| + switch rt := b.row(); { |
| + case b.noRoute: |
| + if rt != nil { |
| + b.t.Errorf("%s with a route at the host: %+v", what, rt) |
| + } |
| + case rt == nil || !reflect.DeepEqual(*rt, b.original): |
| + b.t.Errorf("%s before the original route was back: row %+v", what, rt) |
| + } |
| + if j := b.journal(); j != nil && j.State != registry.JournalCleanupPending { |
| + b.t.Errorf("%s with the journal at %s", what, j.State) |
| + } |
| + if len(b.fenced) != 0 { |
| + b.t.Errorf("%s with a fence held: the cleanup never fences", what) |
| + } |
| +} |
| + |
| +func (b *rollbackBox) hooks(h *SeedHooks) { |
| + h.NextFenceGen = func() uint64 { |
| + b.steps = append(b.steps, "gen") |
| + b.gen++ |
| + return b.gen |
| + } |
| + h.FenceForRollback = func(_ context.Context, host string, gen uint64) (bool, error) { |
| + b.steps = append(b.steps, "fence") |
| + if j := b.journal(); j == nil || j.State != registry.JournalRollingBack || j.FenceGen != gen { |
| + b.t.Errorf("fenced with generation %d under journal %+v", gen, j) |
| + } |
| + if rt := b.row(); rt != nil && rt.ImportGate != registry.ImportGateClosed { |
| + b.t.Errorf("fenced before the gate was closed: %+v", rt) |
| + } |
| + if err := b.trip("fence"); err != nil { |
| + return false, err |
| + } |
| + if held, ok := b.fenced[host]; ok { |
| + return held == gen, nil // ours from an earlier pass, or another operation's |
| + } |
| + if !b.proceed { |
| + return false, nil |
| + } |
| + b.fenced[host] = gen |
| + return true, nil |
| + } |
| + h.LiftFence = func(host string, gen uint64) { |
| + b.steps = append(b.steps, "lift") |
| + _ = b.trip("lift") // a lift cannot fail; only a crash stops it |
| + if b.fenced[host] == gen { |
| + delete(b.fenced, host) |
| + } |
| + } |
| + h.RemoveLocal = func(rt registry.Route) error { |
| + b.steps = append(b.steps, "remove-local") |
| + if _, ok := b.fenced[rt.Host]; !ok { |
| + b.t.Errorf("removed local data of a host that is not fenced and stopped") |
| + } |
| + if j := b.journal(); j == nil || j.State != registry.JournalRollingBack { |
| + b.t.Errorf("removed local data with the journal at %+v", j) |
| + } |
| + if rt.ImportGate != registry.ImportGateClosed || rt.DBPath != NestedDBPath("/data", rt.Host) { |
| + b.t.Errorf("removed local data for a row that is not the closed import row: %+v", rt) |
| + } |
| + if err := b.trip("remove-local"); err != nil { |
| + return err |
| + } |
| + b.local = false |
| + return b.trip("remove-local-done") |
| + } |
| + h.DeleteReplica = func(_ context.Context, host string) error { |
| + b.steps = append(b.steps, "delete-replica") |
| + b.wantRestored("deleted the replica") |
| + if err := b.trip("delete-replica"); err != nil { |
| + return err |
| + } |
| + b.replica = false |
| + return nil |
| + } |
| +} |
| + |
| +// faultData is the region data bucket, with the snapshot delete checked and |
| +// faultable. |
| +type faultData struct { |
| + release.Store |
| + box *rollbackBox |
| +} |
| + |
| +func (d faultData) Delete(ctx context.Context, key string) error { |
| + if strings.HasPrefix(key, seeds.Prefix) { |
| + d.box.steps = append(d.box.steps, "delete-snapshot") |
| + d.box.wantRestored("deleted the snapshot") |
| + if err := d.box.trip("delete-snapshot"); err != nil { |
| + return err |
| + } |
| + } |
| + return d.Store.Delete(ctx, key) |
| +} |
| + |
| +// faultRecords is the release bucket, with the record CAS faultable. |
| +type faultRecords struct { |
| + release.Store |
| + box *rollbackBox |
| + key string |
| +} |
| + |
| +func (f faultRecords) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == f.key { |
| + if err := f.box.trip("cas"); err != nil { |
| + return "", err |
| + } |
| + } |
| + return f.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// rollbackScene is an import this box swapped, verified once (its local |
| +// database placed and replicated) and then aborted. |
| +type rollbackScene struct { |
| + raw *release.DirStore |
| + store release.Store |
| + reg *registry.Registry |
| + path string |
| + rec Record |
| + other string // another seed of the host, with its own snapshot |
| + cfg ReconcileConfig |
| + box *rollbackBox |
| +} |
| + |
| +func abortedImport(t *testing.T) *rollbackScene { |
| + t.Helper() |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg, path := regAt(t, addrRow(rec, sqid)) |
| + box := newRollbackBox(t, reg, *mustGet(t, reg, rec.Host)) |
| + cfg := importCfg(t, nil) |
| + box.hooks(cfg.Seeds) |
| + data := cfg.Seeds.Data |
| + cfg.Seeds.Data = faultData{data, box} |
| + other, _ := seeds.NewID() |
| + for _, id := range []string{rec.Seed.ID, other} { |
| + if _, err := data.Put(ctx, seeds.SnapshotKey(rec.Host, id), []byte("db"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + x := &rollbackScene{raw: s, store: faultRecords{s, box, RecordKey(rec.Account, rec.App, rec.Host)}, |
| + reg: reg, path: path, rec: rec, other: other, cfg: cfg, box: box} |
| + x.pass(t) |
| + if got := mustGet(t, reg, rec.Host); !registry.IsImportRow(*got, registry.ImportJournal{Host: rec.Host, SeedID: rec.Seed.ID}) { |
| + t.Fatalf("the swap did not happen: %+v", got) |
| + } |
| + if _, err := SetRecordSeed(ctx, s, rec.Account, rec.App, rec.Host, SeedAbort(rec.Seed.ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + box.steps = nil |
| + return x |
| +} |
| + |
| +// pass is one Reconcile. A fault that crashes the process ends the pass |
| +// where it stands, and the box restarts. |
| +func (x *rollbackScene) pass(t *testing.T) { |
| + t.Helper() |
| + defer func() { |
| + if r := recover(); r != nil { |
| + if _, ok := r.(crashed); !ok { |
| + panic(r) |
| + } |
| + x.box.restart() |
| + } |
| + }() |
| + if err := Reconcile(ctx, x.store, x.reg, x.cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func (x *rollbackScene) journalState(t *testing.T) string { |
| + t.Helper() |
| + j, err := x.reg.ImportJournal(x.rec.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if j == nil { |
| + return "" |
| + } |
| + return j.State |
| +} |
| + |
| +// wantRolledBack is the clean end state: the original route exactly, no |
| +// journal, this seed's snapshot gone and the other kept, no local data, no |
| +// replica, no fence, and the record rolled back. |
| +func (x *rollbackScene) wantRolledBack(t *testing.T) { |
| + t.Helper() |
| + if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) { |
| + t.Fatalf("row = %+v, want the original %+v", got, x.box.original) |
| + } |
| + if st := x.journalState(t); st != "" { |
| + t.Fatalf("journal still at %s", st) |
| + } |
| + data := x.cfg.Seeds.Data |
| + if _, _, err := data.Get(ctx, seeds.SnapshotKey(x.rec.Host, x.rec.Seed.ID)); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("this seed's snapshot: %v", err) |
| + } |
| + if _, _, err := data.Get(ctx, seeds.SnapshotKey(x.rec.Host, x.other)); err != nil { |
| + t.Fatalf("another seed's snapshot: %v", err) |
| + } |
| + if x.box.local || x.box.replica || len(x.box.fenced) != 0 { |
| + t.Fatalf("local %v replica %v fenced %v", x.box.local, x.box.replica, x.box.fenced) |
| + } |
| + if st := seedState(t, x.raw, x.rec); st.ID != x.rec.Seed.ID || st.State != seeds.StateRolledBack { |
| + t.Fatalf("record seed = %+v, want rolled-back", st) |
| + } |
| +} |
| + |
| +func (x *rollbackScene) trigger(t *testing.T, ddl string) { |
| + t.Helper() |
| + addTrigger(t, x.path, ddl) |
| +} |
| + |
| +func count(steps []string, step string) int { |
| + n := 0 |
| + for _, s := range steps { |
| + if s == step { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// Spec §9 and preflight M13: a failure between any two of the rollback's |
| +// steps, as an error the process survives or as a crash that loses the |
| +// in-memory fence, is resumed by the next passes to the same clean end, |
| +// with nothing lost and nothing done twice that must not be: one fence |
| +// generation per close, no local removal or fence once the route is back, |
| +// and no bucket delete before it is. |
| +func TestRollbackResumesAfterAFailureBetweenAnyTwoSteps(t *testing.T) { |
| + const raise = ` BEGIN SELECT RAISE(ABORT, 'injected'); END` |
| + for _, f := range []struct { |
| + name string |
| + hook string // a box hook or a bucket write that fails once |
| + trigger string // or a registry write that fails, until dropped |
| + after string // the journal's state after the failed pass |
| + crashOnly bool |
| + }{ |
| + {name: "closing the gate", trigger: `CREATE TRIGGER fault BEFORE UPDATE OF state ON import_journal WHEN NEW.state = 'rolling-back'` + raise, after: registry.JournalSwapped}, |
| + {name: "fencing", hook: "fence", after: registry.JournalRollingBack}, |
| + {name: "removing the local data", hook: "remove-local", after: registry.JournalRollingBack}, |
| + {name: "after removing the local data", hook: "remove-local-done", after: registry.JournalRollingBack}, |
| + {name: "restoring the route", trigger: `CREATE TRIGGER fault BEFORE UPDATE OF state ON import_journal WHEN NEW.state = 'cleanup-pending'` + raise, after: registry.JournalRollingBack}, |
| + {name: "lifting the fence", hook: "lift", after: registry.JournalCleanupPending, crashOnly: true}, |
| + {name: "deleting the snapshot", hook: "delete-snapshot", after: registry.JournalCleanupPending}, |
| + {name: "deleting the replica", hook: "delete-replica", after: registry.JournalCleanupPending}, |
| + {name: "deleting the journal row", trigger: `CREATE TRIGGER fault BEFORE DELETE ON import_journal` + raise, after: registry.JournalCleanupPending}, |
| + {name: "marking the record rolled back", hook: "cas", after: ""}, |
| + } { |
| + for _, crash := range []bool{false, true} { |
| + if f.crashOnly && !crash { |
| + continue |
| + } |
| + name := f.name + "/error" |
| + if crash { |
| + name = f.name + "/crash" |
| + } |
| + t.Run(name, func(t *testing.T) { |
| + x := abortedImport(t) |
| + x.box.fault, x.box.crash = f.hook, crash |
| + if f.trigger != "" { |
| + x.trigger(t, f.trigger) |
| + } |
| + x.pass(t) |
| + if x.box.fault != "" { |
| + t.Fatalf("the fault at %s was never reached: steps %q", f.hook, x.box.steps) |
| + } |
| + if st := x.journalState(t); st != f.after { |
| + t.Fatalf("after the failed pass the journal is %q, want %q", st, f.after) |
| + } |
| + if f.trigger != "" { |
| + x.trigger(t, `DROP TRIGGER fault`) |
| + if crash { |
| + x.box.restart() |
| + } |
| + } |
| + if f.after == "" { |
| + if st := seedState(t, x.raw, x.rec); st.State != seeds.StateAborted { |
| + t.Fatalf("record = %s, want aborted with no journal row", st.State) |
| + } |
| + } |
| + restoredAt := -1 |
| + for i := 0; i < 3; i++ { |
| + if restoredAt < 0 && x.journalState(t) != registry.JournalSwapped && x.journalState(t) != registry.JournalRollingBack { |
| + restoredAt = len(x.box.steps) |
| + } |
| + x.pass(t) |
| + } |
| + x.wantRolledBack(t) |
| + gens := 1 |
| + if f.after == registry.JournalSwapped { |
| + gens = 2 // the failed close's generation was never stored |
| + } |
| + if n := count(x.box.steps, "gen"); n != gens { |
| + t.Fatalf("minted %d fence generations, want %d: %q", n, gens, x.box.steps) |
| + } |
| + if restoredAt >= 0 { |
| + for _, s := range x.box.steps[restoredAt:] { |
| + if s == "fence" || s == "remove-local" || s == "gen" { |
| + t.Fatalf("%s after the route was restored: %q", s, x.box.steps) |
| + } |
| + } |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// Preflight M13: the record is deleted and the fence defers. The gated row |
| +// is Provisioned and its record gone, which is what the removal phase |
| +// deletes, but the journal owns it: it survives, closed, every pass. |
| +func TestRollbackRecordDeletedFenceDeferredRowSurvivesRemoval(t *testing.T) { |
| + x := abortedImport(t) |
| + if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.box.proceed = false |
| + for i := 0; i < 2; i++ { |
| + x.pass(t) |
| + got := mustGet(t, x.reg, x.rec.Host) |
| + if got.Kind != "instance" || got.ImportGate != registry.ImportGateClosed || !got.Provisioned { |
| + t.Fatalf("pass %d: row = %+v, want the closed import row", i, got) |
| + } |
| + if st := x.journalState(t); st != registry.JournalRollingBack { |
| + t.Fatalf("pass %d: journal = %s", i, st) |
| + } |
| + } |
| + if !x.box.local || !x.box.replica || count(x.box.steps, "remove-local") != 0 { |
| + t.Fatalf("a deferred rollback removed or deleted something: %q", x.box.steps) |
| + } |
| + x.box.proceed = true |
| + x.pass(t) |
| + if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) { |
| + t.Fatalf("row = %+v", got) |
| + } |
| + if x.journalState(t) != "" || x.box.local || x.box.replica { |
| + t.Fatalf("not finished: journal %q local %v replica %v", x.journalState(t), x.box.local, x.box.replica) |
| + } |
| +} |
| + |
| +// The rollback stops the instance before it removes or deletes anything, |
| +// restores the route before the bucket is touched, and the removal phase, |
| +// which runs on the route list read before the rollback, leaves the |
| +// restored legacy row alone (the journaled-host skip): here the record is |
| +// gone, so without the skip the stale Provisioned import row would have |
| +// its host removed, taking the legacy row with it. |
| +func TestRollbackOrderAndTheRemovalPhaseKeepsTheRestoredRow(t *testing.T) { |
| + x := abortedImport(t) |
| + if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.pass(t) |
| + want := []string{"gen", "fence", "remove-local", "lift", "delete-snapshot", "delete-replica"} |
| + if !reflect.DeepEqual(x.box.steps, want) { |
| + t.Fatalf("steps = %q, want %q", x.box.steps, want) |
| + } |
| + if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, x.box.original) { |
| + t.Fatalf("row after the pass = %+v, want the original", got) |
| + } |
| + if x.journalState(t) != "" { |
| + t.Fatalf("journal = %s", x.journalState(t)) |
| + } |
| +} |
| + |
| +// insertJournal writes a journal row straight into the registry file, as |
| +// a cold restore of an older snapshot would bring it back. |
| +func insertJournal(t *testing.T, path string, j registry.ImportJournal) { |
| + t.Helper() |
| + orig, err := json.Marshal(j.Original) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + db, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer db.Close() |
| + if _, err := db.Exec(`INSERT INTO import_journal (host, seed_id, original, record_key, state, fence_gen, at) VALUES (?,?,?,?,?,?,?)`, |
| + j.Host, j.SeedID, string(orig), j.RecordKey, j.State, j.FenceGen, j.At); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// The Task 4 review's stale journal row: the import was rolled back and |
| +// its record finished, the host was created again, and then a journal row |
| +// for the old import comes back (a cold restore of an older snapshot). |
| +// Whatever state the row comes back in, nothing the new instance owns is |
| +// touched: not its row, its record, its process, its local data or its |
| +// replica. |
| +func TestRollbackOfAStaleJournalTouchesNothingTheNewInstanceOwns(t *testing.T) { |
| + for _, owner := range []string{"the same app", "another app"} { |
| + for _, state := range []string{registry.JournalSwapped, registry.JournalRollingBack, registry.JournalCleanupPending} { |
| + t.Run(owner+"/"+state, func(t *testing.T) { |
| + x := abortedImport(t) |
| + stale, _ := x.reg.ImportJournal(x.rec.Host) |
| + x.pass(t) |
| + x.wantRolledBack(t) |
| + // The console finishes A and the host is created again. |
| + if err := DeleteRecord(ctx, x.raw, x.rec.Account, x.rec.App, x.rec.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + next := shopRecord(x.rec.Account) |
| + if owner == "another app" { |
| + next.App = "other" |
| + if err := PutManifest(ctx, x.raw, next.Account, "other", Manifest{Schema: 1, App: "other", Domain: "other.example.com"}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := CreateRecord(ctx, x.raw, next); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The new instance's row, as the box serves it. |
| + if err := x.reg.Delete(next.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.pass(t) |
| + newRow := mustGet(t, x.reg, next.Host) |
| + if newRow.App != next.App || !newRow.Provisioned || newRow.ImportGate != "" { |
| + t.Fatalf("the new instance was not minted: %+v", newRow) |
| + } |
| + x.box.local, x.box.replica = true, true // its own data now |
| + x.box.original = *newRow // what any bucket delete must find at the host |
| + newRec, _ := GetRecord(ctx, x.raw, next.Account, next.App, next.Host) |
| + |
| + j := *stale |
| + j.State = state |
| + if state != registry.JournalSwapped { |
| + j.FenceGen = 7 |
| + } |
| + insertJournal(t, x.path, j) |
| + x.box.steps = nil |
| + logged := captureLog(t) |
| + x.pass(t) |
| + x.pass(t) |
| + |
| + if got := mustGet(t, x.reg, next.Host); !reflect.DeepEqual(*got, *newRow) { |
| + t.Fatalf("the new instance's row changed:\n got %+v\nwant %+v", got, newRow) |
| + } |
| + if got, _ := GetRecord(ctx, x.raw, next.Account, next.App, next.Host); !reflect.DeepEqual(got, newRec) { |
| + t.Fatalf("the new instance's record changed: %+v", got) |
| + } |
| + for _, s := range x.box.steps { |
| + if s != "delete-snapshot" { |
| + t.Fatalf("the stale journal reached the box: %q\n%s", x.box.steps, logged) |
| + } |
| + } |
| + if !x.box.local || !x.box.replica { |
| + t.Fatalf("the new instance's data went: local %v replica %v", x.box.local, x.box.replica) |
| + } |
| + // A stale row that would act on the route is held; one at |
| + // cleanup-pending finishes its own seed's cleanup and goes. |
| + if st := x.journalState(t); (state == registry.JournalCleanupPending) != (st == "") { |
| + t.Fatalf("journal = %q after a stale %s row", st, state) |
| + } |
| + why := "the route at the host is not the import's" |
| + if state == registry.JournalCleanupPending { |
| + why = "the host's replica is left alone" |
| + } |
| + if !strings.Contains(logged.String(), why) { |
| + t.Fatalf("not held for the reason %q:\n%s", why, logged) |
| + } |
| + }) |
| + } |
| + } |
| +} |
| + |
| +// The rollback tolerates a missing import row (the Task 4 review's carry): |
| +// something removed it, so no route is put back, nothing local is removed |
| +// for a row that is not there, and the cleanup carries on. |
| +func TestRollbackOfAMissingRowSkipsTheRestore(t *testing.T) { |
| + x := abortedImport(t) |
| + if err := x.reg.Delete(x.rec.Host); err != nil { |
| + t.Fatal(err) |
| + } |
| + x.box.noRoute = true |
| + x.pass(t) |
| + if rt, _ := x.reg.Get(x.rec.Host); rt != nil { |
| + t.Fatalf("a route came back: %+v", rt) |
| + } |
| + want := []string{"gen", "fence", "lift", "delete-snapshot", "delete-replica"} |
| + if !reflect.DeepEqual(x.box.steps, want) { |
| + t.Fatalf("steps = %q, want %q", x.box.steps, want) |
| + } |
| + if x.journalState(t) != "" || x.box.replica { |
| + t.Fatalf("journal %q replica %v", x.journalState(t), x.box.replica) |
| + } |
| + if st := seedState(t, x.raw, x.rec); st.State != seeds.StateRolledBack { |
| + t.Fatalf("record = %s", st.State) |
| + } |
| +} |
| + |
| +// Step 6 deletes the host's replica only when it can tell the replica is |
| +// this import's: a record it cannot read, a host listing it cannot finish, |
| +// or a record naming this seed in a state that is not abandoned all defer |
| +// the cleanup; the journal stays at cleanup-pending. |
| +func TestRollbackDefersTheReplicaDeleteWhenItCannotTell(t *testing.T) { |
| + for _, c := range []string{"the record read fails", "the record names this seed, placed"} { |
| + t.Run(c, func(t *testing.T) { |
| + x := abortedImport(t) |
| + if err := x.reg.CloseImportGate(x.rec.Host, x.rec.Seed.ID, 9); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := x.reg.RestoreImportRoute(x.rec.Host, x.rec.Seed.ID); err != nil { |
| + t.Fatal(err) |
| + } |
| + key := RecordKey(x.rec.Account, x.rec.App, x.rec.Host) |
| + switch c { |
| + case "the record read fails": |
| + // Only the step-6 re-read fails: the pass's own read works. |
| + x.store = &failNthGet{Store: x.store, key: key, n: 2} |
| + default: |
| + placed := x.rec |
| + seed := *x.rec.Seed |
| + seed.State = seeds.StatePlaced |
| + placed.Seed = &seed |
| + putReplace(t, x.raw, key, mustJSON(t, placed)) |
| + } |
| + logged := captureLog(t) |
| + x.pass(t) |
| + if st := x.journalState(t); st != registry.JournalCleanupPending { |
| + t.Fatalf("journal = %q", st) |
| + } |
| + if !strings.Contains(logged.String(), "cleanup deferred") { |
| + t.Fatalf("not deferred:\n%s", logged) |
| + } |
| + if !x.box.replica || count(x.box.steps, "delete-replica") != 0 { |
| + t.Fatalf("the replica was deleted: %q", x.box.steps) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// failNthGet fails every read of one key from the nth on. |
| +type failNthGet struct { |
| + release.Store |
| + key string |
| + n int |
| + seen int |
| +} |
| + |
| +func (f *failNthGet) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == f.key { |
| + f.seen++ |
| + if f.seen >= f.n { |
| + return nil, "", errors.New("connection reset") |
| + } |
| + } |
| + return f.Store.Get(ctx, key) |
| +} |
| + |
| +// Without every box hook the rollback does nothing at all: the import |
| +// stays journaled, gated and untouched. |
| +func TestRollbackNeedsEveryHook(t *testing.T) { |
| + x := abortedImport(t) |
| + x.cfg.Seeds.DeleteReplica = nil |
| + row := *mustGet(t, x.reg, x.rec.Host) |
| + x.pass(t) |
| + if got := mustGet(t, x.reg, x.rec.Host); !reflect.DeepEqual(*got, row) { |
| + t.Fatalf("row changed: %+v", got) |
| + } |
| + if st := x.journalState(t); st != registry.JournalSwapped || len(x.box.steps) != 0 { |
| + t.Fatalf("journal %q steps %q", st, x.box.steps) |
| + } |
| +} |
| + |
| +// The route is read again under the fence: a row that stopped being the |
| +// import's while the host was being fenced (removed and replaced by |
| +// another, here) is not removed, the fence is lifted, and the rollback |
| +// holds with the journal at rolling-back. |
| +func TestRollbackRereadsTheRouteUnderTheFence(t *testing.T) { |
| + x := abortedImport(t) |
| + other := addrRow(x.rec, x.rec.Account) |
| + other.Addr = "127.0.0.1:9999" |
| + fence := x.cfg.Seeds.FenceForRollback |
| + x.cfg.Seeds.FenceForRollback = func(ctx context.Context, host string, gen uint64) (bool, error) { |
| + proceed, err := fence(ctx, host, gen) |
| + if err := x.reg.Delete(host); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := x.reg.Upsert(other); err != nil { |
| + t.Fatal(err) |
| + } |
| + return proceed, err |
| + } |
| + x.pass(t) |
| + placed := mustGet(t, x.reg, x.rec.Host) |
| + if placed.Addr != other.Addr || placed.ImportGate != "" { |
| + t.Fatalf("the replacing row changed: %+v", placed) |
| + } |
| + if count(x.box.steps, "remove-local") != 0 || !x.box.local || len(x.box.fenced) != 0 { |
| + t.Fatalf("steps %q local %v fenced %v", x.box.steps, x.box.local, x.box.fenced) |
| + } |
| + if st := x.journalState(t); st != registry.JournalRollingBack { |
| + t.Fatalf("journal = %q", st) |
| + } |
| +} |
| + |
| +// Preflight M3: a journal row whose seed id fails seeds.Check names no |
| +// snapshot and no replica of this import: the cleanup deletes nothing in |
| +// the bucket and only retires the journal row. |
| +func TestRollbackCleanupOfAnInvalidSeedIDDeletesNothing(t *testing.T) { |
| + s, sqid := shopSetup(t) |
| + rec := importRecord(t, s, sqid) |
| + reg, path := regAt(t, addrRow(rec, sqid)) |
| + original := *mustGet(t, reg, rec.Host) |
| + insertJournal(t, path, registry.ImportJournal{Host: rec.Host, SeedID: "../not-a-seed", Original: original, |
| + RecordKey: RecordKey(rec.Account, rec.App, rec.Host), State: registry.JournalCleanupPending, FenceGen: 3, At: 1}) |
| + cfg := importCfg(t, nil) |
| + b := newRollbackBox(t, reg, original) |
| + b.hooks(cfg.Seeds) |
| + cfg.Seeds.Data = faultData{cfg.Seeds.Data, b} |
| + if err := Reconcile(ctx, s, reg, cfg); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(b.steps) != 0 || !b.replica { |
| + t.Fatalf("steps %q replica %v", b.steps, b.replica) |
| + } |
| + if j, _ := reg.ImportJournal(rec.Host); j != nil { |
| + t.Fatalf("journal = %+v", j) |
| + } |
| + if got := mustGet(t, reg, rec.Host); !reflect.DeepEqual(*got, original) { |
| + t.Fatalf("row = %+v", got) |
| + } |
| +} |
amadan