amadan

carlos / platform Public

registry, domains, activator: the rollback acts only on the import's row (Plan 7 Task 5, part 1)

- registry.IsImportRow: the row an import's swap made (instance, provisioned,
  gated, its seed column naming the journal's seed). CloseImportGate and
  RestoreImportRoute refuse any other row at the host and change nothing,
  so a stale journal row from a snapshot never gates or replaces a later
  instance's row. A host with no row is closed and moved to cleanup-pending
  with no route put back.
- SwapForImport refuses an empty record key (Task 4 N1) and a next row
  IsImportRow would not recognise.
- domains.Reconcile's trash removal skips hosts in the import journal, as
  the provision pass's removal does; an unreadable journal removes no
  trashed route that pass.
- activator.PurgeSeedLocal exports Plan 4's purgeSeedTarget for the
  rollback's local removal (preflight M12): the database, -wal, -shm,
  <db>.seed, litestream metadata and .seed-sd_*.tmp, through instancefs.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude dcb60d715bc76e17dda81b49e37b95b7cb089ca8 parent 44284ba
7 files changed, +488 −44
  • internal/activator/seedplace.go +21 −0
  • internal/activator/seedpurge_linux_test.go +86 −0
  • internal/domains/importjournal_test.go +106 −0
  • internal/domains/reconcile.go +33 −0
  • internal/provision/seedswap_test.go +10 −8
  • internal/registry/importjournal.go +71 −8
  • internal/registry/importjournal_test.go +161 −28
diff --git a/internal/activator/seedplace.go b/internal/activator/seedplace.go
index 818213ed2..5ef20a9cb 100644
--- a/internal/activator/seedplace.go
+++ b/internal/activator/seedplace.go
@@ -220,6 +220,27 @@ func notAPlacedFile(err error) bool {
// directory. The id is a seed id (seeds.Check), one clean path element.
func seedTempName(id string) string { return ".seed-" + id + ".tmp" }
+// PurgeSeedLocal removes an instance's local database at dbPath and
+// everything beside it that could stand in for one (purgeSeedTarget): the
+// database, -wal, -shm, the <db>.seed marker, litestream's metadata and
+// any seed fetch leftovers, through instancefs. It is the import
+// rollback's local removal (seeded-instances spec §5.6 step 3), run after
+// the host is fenced and stopped. A directory that is not there holds
+// nothing, so a second run, or one for an instance never woken, is done.
+func PurgeSeedLocal(dbPath string) error {
+ if !filepath.IsAbs(dbPath) || filepath.Clean(dbPath) != dbPath {
+ return fmt.Errorf("activator: %q is not a clean absolute database path", dbPath)
+ }
+ dir, base := filepath.Split(dbPath)
+ if base == "" {
+ return fmt.Errorf("activator: %q names no database", dbPath)
+ }
+ if _, err := os.Lstat(dir); errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ return purgeSeedTarget(filepath.Clean(dir), base)
+}
+
// purgeSeedTarget removes everything at and beside dir/base that could
// stand in for a seed's database: what PurgeLocalDBArtifacts removes in its
// full form, the marker, and any seed temp file, through instancefs.
diff --git a/internal/activator/seedpurge_linux_test.go b/internal/activator/seedpurge_linux_test.go
new file mode 100644
index 000000000..de228fe04
--- /dev/null
+++ b/internal/activator/seedpurge_linux_test.go
@@ -0,0 +1,86 @@
+//go:build linux
+
+package activator
+
+import (
+ "os"
+ "path/filepath"
+ "testing"
+)
+
+// PurgeSeedLocal is the import rollback's local removal (preflight M12):
+// Plan 4's purge, so it covers the seed fetch leftovers (a copy of the
+// customer database) and goes through instancefs. It leaves anything else
+// in the directory, is idempotent, and is done for a directory that is
+// not there.
+func TestPurgeSeedLocalRemovesTheDatabaseAndEverythingThatStandsInForIt(t *testing.T) {
+ dir := filepath.Join(t.TempDir(), "instances", "shop")
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ db := filepath.Join(dir, "a.shop.example.com.db")
+ gone := []string{db, db + "-wal", db + "-shm", db + SeedMarkerSuffix,
+ filepath.Join(dir, ".seed-sd_aaaaaaaaaaaaaaaaaaaaaaaaaa.tmp")}
+ for _, p := range gone {
+ if err := os.WriteFile(p, []byte("x"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, d := range []string{db + "-litestream", filepath.Join(dir, ".activator-a.shop.example.com.db-litestream")} {
+ if err := os.MkdirAll(filepath.Join(d, "generations"), 0o700); err != nil {
+ t.Fatal(err)
+ }
+ gone = append(gone, d)
+ }
+ kept := filepath.Join(dir, "uploads")
+ if err := os.WriteFile(kept, []byte("keep"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < 2; i++ {
+ if err := PurgeSeedLocal(db); err != nil {
+ t.Fatalf("run %d: %v", i, err)
+ }
+ }
+ for _, p := range gone {
+ if _, err := os.Lstat(p); !os.IsNotExist(err) {
+ t.Errorf("%s survived", p)
+ }
+ }
+ if _, err := os.Stat(kept); err != nil {
+ t.Fatalf("an unrelated file went: %v", err)
+ }
+ if err := PurgeSeedLocal(filepath.Join(t.TempDir(), "missing", "a.shop.example.com.db")); err != nil {
+ t.Fatalf("a missing directory: %v", err)
+ }
+ for _, bad := range []string{"a.shop.example.com.db", "/data/instances/x/../a.db", "/data/instances/x/"} {
+ if err := PurgeSeedLocal(bad); err == nil {
+ t.Errorf("%q: no error", bad)
+ }
+ }
+}
+
+// A symlink at the database's name is removed as a link; its target stays.
+func TestPurgeSeedLocalNeverFollowsASymlink(t *testing.T) {
+ root := t.TempDir()
+ dir := filepath.Join(root, "instance")
+ if err := os.MkdirAll(dir, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ target := filepath.Join(root, "elsewhere.db")
+ if err := os.WriteFile(target, []byte("someone else's"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ db := filepath.Join(dir, "a.shop.example.com.db")
+ if err := os.Symlink(target, db); err != nil {
+ t.Fatal(err)
+ }
+ if err := PurgeSeedLocal(db); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := os.Lstat(db); !os.IsNotExist(err) {
+ t.Fatal("the link survived")
+ }
+ if b, err := os.ReadFile(target); err != nil || string(b) != "someone else's" {
+ t.Fatalf("the link's target was touched: %q %v", b, err)
+ }
+}
diff --git a/internal/domains/importjournal_test.go b/internal/domains/importjournal_test.go
new file mode 100644
index 000000000..cec4ec1eb
--- /dev/null
+++ b/internal/domains/importjournal_test.go
@@ -0,0 +1,106 @@
+package domains_test
+
+import (
+ "database/sql"
+ "path/filepath"
+ "reflect"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// trashedImportScene is a trashed app "shop" with two routes on this box:
+// a.shop.example.com, swapped for an uncommitted import (so journaled),
+// and b.shop.example.com, an ordinary route.
+func trashedImportScene(t *testing.T) (*release.DirStore, *registry.Registry, string, registry.Route) {
+ t.Helper()
+ s := dirStore(t)
+ sqid := writeAccountManifest(t, s, 793481920157, "Shop")
+ putKey(t, s, "apps/"+sqid+"/shop/channels/stable.json")
+ path := filepath.Join(t.TempDir(), "registry.db")
+ reg, err := registry.Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ legacy := registry.Route{Host: "a.shop.example.com", Kind: "service", Addr: "127.0.0.1:8401", App: "shop",
+ Account: sqid, Status: "ready", Channel: "stable", CreatedAt: 1700000000}
+ plain := legacy
+ plain.Host, plain.Addr = "b.shop.example.com", "127.0.0.1:8402"
+ for _, rt := range []registry.Route{legacy, plain} {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ original := *mustGet(t, reg, legacy.Host)
+ const seedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
+ next := registry.Route{Host: legacy.Host, Kind: "instance", Socket: "/run/carlos/instances/a/app.sock",
+ DBPath: "/data/instances/a/a.shop.example.com.db", App: "shop", Account: sqid, Status: "ready", Channel: "stable",
+ Hibernate: true, Provisioned: true, ImportGate: "abc123",
+ SeedSource: `{"id":"` + seedID + `","kind":"import","state":"swapped"}`}
+ if err := reg.SwapForImport(registry.ImportJournal{Host: legacy.Host, SeedID: seedID, Original: original,
+ RecordKey: "apps/" + sqid + "/shop/instances/a.shop.example.com.json", At: 1}, next); err != nil {
+ t.Fatal(err)
+ }
+ if err := release.TrashApp(ctx, s, sqid, "shop", "owner@example.com", time.Unix(1753900000, 0)); err != nil {
+ t.Fatal(err)
+ }
+ return s, reg, path, *mustGet(t, reg, legacy.Host)
+}
+
+// Task 5 (the Task 4 review's carry): the trash removal skips a host the
+// import journal holds, as the provision pass's removal does. The journal
+// owns the gated row until its rollback restores the original route and
+// lets go; every other route of the trashed app goes as before.
+func TestReconcileTrashLeavesAJournaledImportsRow(t *testing.T) {
+ s, reg, _, gated := trashedImportScene(t)
+ for pass := 1; pass <= 2; pass++ {
+ if err := domains.Reconcile(ctx, s, reg, cfg(t)); err != nil {
+ t.Fatalf("pass %d: %v", pass, err)
+ }
+ if got := mustGet(t, reg, gated.Host); !reflect.DeepEqual(*got, gated) {
+ t.Fatalf("pass %d: the journaled row changed: %+v", pass, got)
+ }
+ if rt, _ := reg.Get("b.shop.example.com"); rt != nil {
+ t.Fatalf("pass %d: the trashed app's other route survived: %+v", pass, rt)
+ }
+ }
+ // Once the journal lets go, the trash removes the host like any other.
+ if err := reg.CloseImportGate(gated.Host, "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa", 1); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.RestoreImportRoute(gated.Host, "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.DeleteImportJournal(gated.Host, "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"); err != nil {
+ t.Fatal(err)
+ }
+ if err := domains.Reconcile(ctx, s, reg, cfg(t)); err != nil {
+ t.Fatal(err)
+ }
+ if rt, _ := reg.Get(gated.Host); rt != nil {
+ t.Fatalf("the trashed app's restored route survived: %+v", rt)
+ }
+}
+
+// A journal that cannot be read could hide any host: no route of a
+// trashed app is removed that pass.
+func TestReconcileTrashRemovesNothingWhenTheJournalCannotBeRead(t *testing.T) {
+ s, reg, path, gated := trashedImportScene(t)
+ db, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Exec(`UPDATE import_journal SET original = '{' WHERE host = ?`, gated.Host); err != nil {
+ t.Fatal(err)
+ }
+ db.Close()
+ if err := domains.Reconcile(ctx, s, reg, cfg(t)); err != nil {
+ t.Fatal(err)
+ }
+ mustGet(t, reg, gated.Host)
+ mustGet(t, reg, "b.shop.example.com")
+}
diff --git a/internal/domains/reconcile.go b/internal/domains/reconcile.go
index 3a641c12a..9e0488db3 100644
--- a/internal/domains/reconcile.go
+++ b/internal/domains/reconcile.go
@@ -201,10 +201,30 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// Trashed apps stop routing NOW: every route of a trashed (account, app)
// is removed, and the shrunken slice is what minting sees, so nothing
// below re-mints an apex or canary host for an app in the trash.
+ //
+ // Except a host in the import journal (seeded-instances spec §5.7): its
+ // row is an uncommitted import's, and the journal owns it until the
+ // rollback puts the original route back and retires the journal row.
+ // Deleting it here would lose the import's gate and leave the rollback
+ // no row to restore over. The provision pass skips these hosts for the
+ // same reason. The row goes on a later pass, once the journal lets go.
+ // A journal that cannot be read could hide any host, so then no trashed
+ // route is removed this pass.
if len(trashed) > 0 {
+ journaled, jerr := importJournalHosts(reg)
+ if jerr != nil {
+ log.Printf("domains: the import journal could not be read: %v (no route of a trashed app is removed this pass)", jerr)
+ }
keptLive := routes[:0]
for _, rt := range routes {
if rt.App != "" && trashed[appKey{release.RouteAccount(rt), rt.App}] != nil {
+ if jerr != nil || journaled[rt.Host] {
+ if jerr == nil {
+ log.Printf("domains: not removing route %s of trashed app %s yet: an import on it is still journaled", rt.Host, rt.App)
+ }
+ keptLive = append(keptLive, rt)
+ continue
+ }
if err := reg.Delete(rt.Host); err != nil {
log.Printf("domains: remove route %s of trashed app %s: %v", rt.Host, rt.App, err)
keptLive = append(keptLive, rt) // still there; try next pass
@@ -932,3 +952,16 @@ func certExpiry(path string) int64 {
}
return 0
}
+
+// importJournalHosts is every host the registry's import journal holds.
+func importJournalHosts(reg *registry.Registry) (map[string]bool, error) {
+ js, err := reg.ImportJournals()
+ if err != nil {
+ return nil, err
+ }
+ hosts := make(map[string]bool, len(js))
+ for _, j := range js {
+ hosts[j.Host] = true
+ }
+ return hosts, nil
+}
diff --git a/internal/provision/seedswap_test.go b/internal/provision/seedswap_test.go
index 6dd803c58..40016bed7 100644
--- a/internal/provision/seedswap_test.go
+++ b/internal/provision/seedswap_test.go
@@ -542,23 +542,25 @@ func TestSwapResumeNeedsTheRecordKey(t *testing.T) {
}
// A journal row this record's seed does not own (another seed's, or one
-// already rolling back) leaves the record alone: no refusal, no swap.
+// already rolling back) leaves the record alone: no refusal, no swap. This
+// is the swap's own backstop (seedswap.go), called directly: in a pass the
+// journal pass decides such a row first, and rolls it back.
func TestSwapLeavesAnotherSeedsJournalAlone(t *testing.T) {
s, sqid := shopSetup(t)
rec := importRecord(t, s, sqid)
reg := regWith(t, addrRow(rec, sqid))
original := mustGet(t, reg, rec.Host)
- other, _ := seeds.NewID()
- next := routeFor(rec, "/data", "", true)
- next.ImportGate = rec.Seed.VerifyHash
- if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: other, Original: *original, At: 1}, next); err != nil {
+ other := rec
+ seed := *rec.Seed
+ seed.ID, _ = seeds.NewID()
+ other.Seed = &seed
+ if err := reg.SwapForImport(registry.ImportJournal{Host: rec.Host, SeedID: seed.ID, Original: *original,
+ RecordKey: RecordKey(rec.Account, rec.App, rec.Host), At: 1}, importRow(other, "/data")); err != nil {
t.Fatal(err)
}
row := mustGet(t, reg, rec.Host)
j, _ := reg.ImportJournal(rec.Host)
- if err := Reconcile(ctx, s, reg, importCfg(t, nil)); err != nil {
- t.Fatal(err)
- }
+ swapImport(ctx, s, reg, importCfg(t, nil), rec, row, func() (bool, error) { return true, nil })
if st := seedState(t, s, rec); st.State != seeds.StateReady || st.Reason != "" {
t.Fatalf("record = %s %q, want ready", st.State, st.Reason)
}
diff --git a/internal/registry/importjournal.go b/internal/registry/importjournal.go
index cec53db5f..2c1124dcd 100644
--- a/internal/registry/importjournal.go
+++ b/internal/registry/importjournal.go
@@ -174,15 +174,45 @@ func insertRouteTx(tx txer, t Route) error {
return err
}
+// IsImportRow reports whether rt is the row j's swap put in place: the
+// journal's host, an instance row a record provisioned, still gated (by
+// the verify hash, or the closed sentinel once a rollback began), and
+// whose seed column names j's seed. The rollback acts only on this row
+// (spec §5.6): any other row at the host belongs to someone else, such as
+// a later instance when a stale journal row came back from a snapshot.
+func IsImportRow(rt Route, j ImportJournal) bool {
+ return j.SeedID != "" && rt.Host == j.Host && rt.Kind == "instance" && rt.Provisioned &&
+ rt.ImportGate != "" && seedColumnID(rt.SeedSource) == j.SeedID
+}
+
+// seedColumnID is the seed id a seed_source column names, "" for none or
+// one that does not parse. The column is seeds.RowSeed's JSON; the
+// registry reads only its id, since seeds imports this package.
+func seedColumnID(col string) string {
+ if col == "" {
+ return ""
+ }
+ var v struct {
+ ID string `json:"id"`
+ }
+ if err := json.Unmarshal([]byte(col), &v); err != nil {
+ return ""
+ }
+ return v.ID
+}
+
// SwapForImport is the swap's one transaction (spec §5.4, codex 14): with
// no journal row for the host and the stored route still exactly
// j.Original, it replaces the route with next (Upsert's defaults applied)
-// and writes the journal row at swapped. next must be the same host and
-// carry its gate. The original must be one the journal can put back: not
-// claim-owned (claims are off for imports, spec §3.4), not held by an
-// operation, not itself gated, and unchanged by its JSON round trip.
+// and writes the journal row at swapped. next must be the import's row as
+// IsImportRow reads it (the same host, gated, naming the journal's seed),
+// and the journal must name its record: a journal row with no record key
+// can never be decided, so it is never written. The original must be one
+// the journal can put back: not claim-owned (claims are off for imports,
+// spec §3.4), not held by an operation, not itself gated, and unchanged by
+// its JSON round trip.
func (r *Registry) SwapForImport(j ImportJournal, next Route) error {
- if j.Host == "" || j.SeedID == "" || next.Host != j.Host || j.Original.Host != j.Host || next.ImportGate == "" {
+ if j.Host == "" || j.SeedID == "" || j.RecordKey == "" || j.Original.Host != j.Host || !IsImportRow(normalizeRoute(next), j) {
return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host)
}
if j.Original.ClaimOwnership != (ClaimOwnership{}) || j.Original.HoldOp != "" || j.Original.ImportGate != "" {
@@ -239,18 +269,38 @@ func journalFor(tx txer, host, seedID string, states ...string) (*ImportJournal,
return nil, fmt.Errorf("%w: %s's journal is %s", ErrImportJournal, host, j.State)
}
+// importRowTx is host's row when it is j's import row, nil when the host
+// has no row, and ErrImportJournal when the row there is anyone else's.
+func importRowTx(tx txer, j *ImportJournal) (*Route, error) {
+ cur, err := getRouteTx(tx, j.Host)
+ if err != nil {
+ return nil, err
+ }
+ if cur != nil && !IsImportRow(*cur, *j) {
+ return nil, fmt.Errorf("%w: the route at %s is not import %s's", ErrImportJournal, j.Host, j.SeedID)
+ }
+ return cur, nil
+}
+
// CloseImportGate is the rollback's step 1 (spec §5.6): the journal says
// rolling-back with the fence generation, and the gate refuses even verify
// requests, in one transaction. Run again on a row already rolling-back, it
// re-closes the gate and keeps the stored generation (fenceGen is ignored):
// step 2 recognises this rollback's own fence by that generation, so the
// caller reads FenceGen back from the journal rather than trusting its own.
+//
+// It refuses when the host's row is not the import's (IsImportRow), so it
+// never gates a row the import does not own. A host with no row is closed
+// all the same: the journal moves on, and there is no gate to set.
func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error {
return r.inTx(func(tx txer) error {
j, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack)
if err != nil {
return err
}
+ if _, err := importRowTx(tx, j); err != nil {
+ return err
+ }
if j.State == JournalSwapped {
if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil {
return err
@@ -264,17 +314,30 @@ func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error {
// RestoreImportRoute is the rollback's step 4 (spec §5.6, codex R3-1): the
// original route back exactly as it was stored, and the journal at
// cleanup-pending.
+//
+// Only the import's own row is replaced (IsImportRow). Any other row at
+// the host is refused, and nothing changes: it is not the import's to
+// remove. A host with no row at all (something removed the import's row)
+// gets no route back, since whatever removed it decided the host should
+// not be served; the journal still moves to cleanup-pending, so the
+// rollback carries on with its cleanup.
func (r *Registry) RestoreImportRoute(host, seedID string) error {
return r.inTx(func(tx txer) error {
j, err := journalFor(tx, host, seedID, JournalRollingBack)
if err != nil {
return err
}
- if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, host); err != nil {
+ cur, err := importRowTx(tx, j)
+ if err != nil {
return err
}
- if err := insertRouteTx(tx, j.Original); err != nil {
- return err
+ if cur != nil {
+ if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, host); err != nil {
+ return err
+ }
+ if err := insertRouteTx(tx, j.Original); err != nil {
+ return err
+ }
}
_, err = tx.Exec(`UPDATE import_journal SET state = ? WHERE host = ?`, JournalCleanupPending, host)
return err
diff --git a/internal/registry/importjournal_test.go b/internal/registry/importjournal_test.go
index 72bdf3800..dd9d228ea 100644
--- a/internal/registry/importjournal_test.go
+++ b/internal/registry/importjournal_test.go
@@ -11,8 +11,9 @@ import (
// Neutral fixtures: app shop, hosts under shop.example.com. "bab" is the
// pinned ops account (Sqid(1)); "bbw" is Sqid(42), an ordinary account.
const (
- importHost = "old.shop.example.com"
- importSeedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
+ importHost = "old.shop.example.com"
+ importSeedID = "sd_aaaaaaaaaaaaaaaaaaaaaaaaaa"
+ importRecordKey = "apps/bbw/shop/instances/old.shop.example.com.json"
)
func legacyRow() Route {
@@ -26,6 +27,14 @@ func execRow() Route {
Channel: "stable", Hibernate: true, Provisioned: true, ImportGate: "abc123", CreatedAt: 1700000001}
}
+// swapRow is execRow as the swap writes it: its seed column names the
+// journal's seed (IsImportRow).
+func swapRow() Route {
+ r := execRow()
+ r.SeedSource = `{"id":"` + importSeedID + `","kind":"import","state":"swapped"}`
+ return r
+}
+
func mustJournal(t *testing.T, reg *Registry, host string) *ImportJournal {
t.Helper()
j, err := reg.ImportJournal(host)
@@ -51,8 +60,8 @@ func TestSwapCommitAndRestore(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, old.Host)
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: "apps/bbw/shop/instances/old.shop.example.com.json", Original: stored, At: 1}
- if err := reg.SwapForImport(j, execRow()); err != nil {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
+ if err := reg.SwapForImport(j, swapRow()); err != nil {
t.Fatal(err)
}
got := mustRoute(t, reg, old.Host)
@@ -64,7 +73,7 @@ func TestSwapCommitAndRestore(t *testing.T) {
t.Fatalf("journal = %+v", jr)
}
// A second swap for the host is refused: the journal is there.
- if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {
+ if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
t.Fatalf("second swap: %v", err)
}
// Close the gate, restore the route exactly, and the journal says where it is.
@@ -126,8 +135,8 @@ func TestCommitImportClearsTheGateAndTheJournal(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, old.Host)
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}
- if err := reg.SwapForImport(j, execRow()); err != nil {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
+ if err := reg.SwapForImport(j, swapRow()); err != nil {
t.Fatal(err)
}
if err := reg.CommitImport(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {
@@ -144,7 +153,7 @@ func TestCommitImportClearsTheGateAndTheJournal(t *testing.T) {
t.Fatalf("after commit: gate %q journal %+v fingerprint moved %v", got.ImportGate, jr, before != after)
}
// The committed row is the exec row, ungated.
- want := execRow()
+ want := swapRow()
want.ImportGate = ""
if !reflect.DeepEqual(got, want) {
t.Fatalf("committed row %+v, want %+v", got, want)
@@ -163,8 +172,8 @@ func TestSwapForImportRefusesAChangedRow(t *testing.T) {
if err := reg.Upsert(changed); err != nil {
t.Fatal(err)
}
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}
- if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
+ if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
t.Fatalf("swap over a changed row: %v", err)
}
if got := mustRoute(t, reg, old.Host); got.Addr != "127.0.0.1:9999" {
@@ -178,8 +187,8 @@ func TestSwapForImportRefusesAChangedRow(t *testing.T) {
func TestSwapForImportRefusesAMissingRowAndAMalformedSwap(t *testing.T) {
reg := openTestRegistry(t)
old := legacyRow()
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: old}
- if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: old}
+ if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
t.Fatalf("swap with no stored row: %v", err)
}
if err := reg.Upsert(old); err != nil {
@@ -187,19 +196,29 @@ func TestSwapForImportRefusesAMissingRowAndAMalformedSwap(t *testing.T) {
}
stored := mustRoute(t, reg, old.Host)
j.Original = stored
- ungated := execRow()
+ ungated := swapRow()
ungated.ImportGate = ""
- other := execRow()
+ other := swapRow()
other.Host = "new.shop.example.com"
noID := j
noID.SeedID = ""
+ noKey := j
+ noKey.RecordKey = ""
+ otherSeed := swapRow()
+ otherSeed.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb","kind":"import","state":"swapped"}`
+ unprovisioned := swapRow()
+ unprovisioned.Provisioned = false
for name, c := range map[string]struct {
j ImportJournal
next Route
}{
- "an ungated row": {j, ungated},
- "another host": {j, other},
- "no seed id": {noID, execRow()},
+ "an ungated row": {j, ungated},
+ "another host": {j, other},
+ "no seed id": {noID, swapRow()},
+ "no record key (N1)": {noKey, swapRow()},
+ "a row naming another seed": {j, otherSeed},
+ "a row with no seed column": {j, execRow()},
+ "a row no record provisioned": {j, unprovisioned},
} {
if err := reg.SwapForImport(c.j, c.next); !errors.Is(err, ErrImportJournal) {
t.Errorf("%s: %v", name, err)
@@ -244,9 +263,9 @@ func TestImportJournalsListsEveryRow(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, host)
- next := execRow()
+ next := swapRow()
next.Host = host
- j := ImportJournal{Host: host, SeedID: importSeedID, Original: stored, At: int64(i + 1)}
+ j := ImportJournal{Host: host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: int64(i + 1)}
if err := reg.SwapForImport(j, next); err != nil {
t.Fatal(err)
}
@@ -400,8 +419,8 @@ func TestSnapshotFingerprintSeesTheJournal(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, old.Host)
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored, At: 1}
- if err := reg.SwapForImport(j, execRow()); err != nil {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
+ if err := reg.SwapForImport(j, swapRow()); err != nil {
t.Fatal(err)
}
if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
@@ -452,8 +471,8 @@ func TestSwapForImportRefusesAnOwnedHeldOrGatedOriginal(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, old.Host)
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}
- if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
+ if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
t.Fatalf("swap: %v", err)
}
if got := mustRoute(t, reg, old.Host); got != stored {
@@ -479,8 +498,8 @@ func TestSwapForImportRefusesAnOriginalJSONWouldChange(t *testing.T) {
if stored.LastError != old.LastError {
t.Skipf("the driver did not keep the raw bytes: %q", stored.LastError)
}
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}
- if err := reg.SwapForImport(j, execRow()); !errors.Is(err, ErrImportJournal) {
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
+ if err := reg.SwapForImport(j, swapRow()); !errors.Is(err, ErrImportJournal) {
t.Fatalf("swap: %v", err)
}
if got := mustRoute(t, reg, old.Host); got != stored {
@@ -497,14 +516,14 @@ func TestJournalTransactionsAreAtomic(t *testing.T) {
t.Fatal(err)
}
stored := mustRoute(t, reg, old.Host)
- j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: stored}
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored}
// The swap: the route DELETE and the exec row INSERT land before the
// journal INSERT fails.
if _, err := reg.sql.Exec(`CREATE TRIGGER fail_swap BEFORE INSERT ON import_journal BEGIN SELECT RAISE(ABORT, 'injected'); END`); err != nil {
t.Fatal(err)
}
- if err := reg.SwapForImport(j, execRow()); err == nil {
+ if err := reg.SwapForImport(j, swapRow()); err == nil {
t.Fatal("swap with a failing journal write: no error")
}
if got := mustRoute(t, reg, old.Host); got != stored {
@@ -516,7 +535,7 @@ func TestJournalTransactionsAreAtomic(t *testing.T) {
if _, err := reg.sql.Exec(`DROP TRIGGER fail_swap`); err != nil {
t.Fatal(err)
}
- if err := reg.SwapForImport(j, execRow()); err != nil {
+ if err := reg.SwapForImport(j, swapRow()); err != nil {
t.Fatal(err)
}
if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
@@ -547,3 +566,117 @@ func TestJournalTransactionsAreAtomic(t *testing.T) {
t.Fatalf("restored %+v, want %+v", got, stored)
}
}
+
+// swappedRegistry is a registry where the import has swapped legacyRow for
+// swapRow, and the original as stored.
+func swappedRegistry(t *testing.T) (*Registry, Route) {
+ t.Helper()
+ reg := openTestRegistry(t)
+ if err := reg.Upsert(legacyRow()); err != nil {
+ t.Fatal(err)
+ }
+ stored := mustRoute(t, reg, importHost)
+ j := ImportJournal{Host: importHost, SeedID: importSeedID, RecordKey: importRecordKey, Original: stored, At: 1}
+ if err := reg.SwapForImport(j, swapRow()); err != nil {
+ t.Fatal(err)
+ }
+ return reg, stored
+}
+
+// Task 5 (a stale journal row from a snapshot): the rollback's two route
+// transactions act only on the import's own row. A row there that is not
+// it (a later instance at the host, the original put back by hand, a row
+// naming another seed or no longer gated) is refused and left exactly as
+// it is, and the journal does not move.
+func TestRollbackTransactionsRefuseARowThatIsNotTheImports(t *testing.T) {
+ later := execRow()
+ later.ImportGate = ""
+ later.Account = "bbx"
+ for name, foreign := range map[string]Route{
+ "a later instance": later,
+ "the original back": legacyRow(),
+ "another seed's row": func() Route { r := swapRow(); r.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"}`; return r }(),
+ "its gate cleared": func() Route { r := swapRow(); r.ImportGate = ""; return r }(),
+ } {
+ t.Run(name, func(t *testing.T) {
+ for _, closeFirst := range []bool{false, true} {
+ reg, _ := swappedRegistry(t)
+ if closeFirst {
+ if err := reg.CloseImportGate(importHost, importSeedID, 7); err != nil {
+ t.Fatal(err)
+ }
+ }
+ before := mustJournal(t, reg, importHost)
+ if _, err := reg.sql.Exec(`DELETE FROM routes WHERE host = ?`, importHost); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(foreign); err != nil {
+ t.Fatal(err)
+ }
+ placed := mustRoute(t, reg, importHost)
+ if err := reg.CloseImportGate(importHost, importSeedID, 8); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("close over %s: %v", name, err)
+ }
+ if closeFirst {
+ if err := reg.RestoreImportRoute(importHost, importSeedID); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("restore over %s: %v", name, err)
+ }
+ }
+ if got := mustRoute(t, reg, importHost); got != placed {
+ t.Fatalf("the row changed: %+v, want %+v", got, placed)
+ }
+ if jr := mustJournal(t, reg, importHost); !reflect.DeepEqual(jr, before) {
+ t.Fatalf("the journal moved: %+v -> %+v", before, jr)
+ }
+ }
+ })
+ }
+}
+
+// Task 5: an import whose row is gone (something removed it) is closed and
+// moved to cleanup-pending with no route put back.
+func TestRollbackTransactionsTolerateAMissingRow(t *testing.T) {
+ reg, _ := swappedRegistry(t)
+ if err := reg.Delete(importHost); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.CloseImportGate(importHost, importSeedID, 7); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.RestoreImportRoute(importHost, importSeedID); err != nil {
+ t.Fatal(err)
+ }
+ if rt, err := reg.Get(importHost); err != nil || rt != nil {
+ t.Fatalf("a route came back for a removed row: %+v, %v", rt, err)
+ }
+ if jr := mustJournal(t, reg, importHost); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 7 {
+ t.Fatalf("journal = %+v", jr)
+ }
+}
+
+func TestIsImportRow(t *testing.T) {
+ j := ImportJournal{Host: importHost, SeedID: importSeedID}
+ closed := swapRow()
+ closed.ImportGate = ImportGateClosed
+ if !IsImportRow(swapRow(), j) || !IsImportRow(closed, j) {
+ t.Fatal("the swap's row, gated or closed, is not recognised")
+ }
+ for name, mutate := range map[string]func(*Route){
+ "another host": func(r *Route) { r.Host = "new.shop.example.com" },
+ "a service": func(r *Route) { r.Kind = "service" },
+ "not provisioned": func(r *Route) { r.Provisioned = false },
+ "ungated": func(r *Route) { r.ImportGate = "" },
+ "no seed column": func(r *Route) { r.SeedSource = "" },
+ "a bad seed column": func(r *Route) { r.SeedSource = "{" },
+ "another seed": func(r *Route) { r.SeedSource = `{"id":"sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"}` },
+ } {
+ r := swapRow()
+ mutate(&r)
+ if IsImportRow(r, j) {
+ t.Errorf("%s: recognised as the import's row", name)
+ }
+ }
+ if IsImportRow(swapRow(), ImportJournal{Host: importHost}) {
+ t.Error("a journal with no seed id recognised a row")
+ }
+}