amadan

carlos / platform Public

registry: the import gate column and the import journal's transactions (Plan 7 Task 1)

import_gate on routes (Upsert inserts it, its conflict path keeps it; folded
into both fingerprints only when set) and the import_journal table with
SwapForImport, CloseImportGate, RestoreImportRoute, CommitImport and
DeleteImportJournal, each one BEGIN IMMEDIATE transaction. Every route insert
binds through routeArgs, the cold handover's included (preflight M6); a
restored original is inserted raw (m4). OpsAccountSqid dropped (M5).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f20d2c5e25f0814baee66304382cfee80835077d parent b7b33e7
5 files changed, +701 −57
  • internal/registry/coldhandover.go +1 −4
  • internal/registry/importjournal.go +262 −0
  • internal/registry/importjournal_test.go +340 −0
  • internal/registry/registry.go +97 −53
  • internal/registry/registry_test.go +1 −0
diff --git a/internal/registry/coldhandover.go b/internal/registry/coldhandover.go
index 980afd118..469ef577a 100644
--- a/internal/registry/coldhandover.go
+++ b/internal/registry/coldhandover.go
@@ -6,7 +6,6 @@ import (
"database/sql"
"encoding/json"
"errors"
- "strings"
)
type ColdRows struct {
@@ -101,9 +100,7 @@ func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string,
if count != 0 {
return ErrClaimOwnership
}
- t := next.Route
- values := []any{t.Host, t.Slug, t.Kind, t.Socket, t.Addr, t.Status, t.LastError, t.DBPath, t.Version, t.PrevVersion, t.App, t.Pool, t.Region, t.Channel, b2iDB(t.Hibernate), b2iDB(t.Canary), t.Replication, t.NextDue, t.Backing, b2iDB(t.StatusNamed), b2iDB(t.Platform), t.DNSStatus, t.DNSCheckedAt, t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel, b2iDB(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health, t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds, t.Replicas, b2iDB(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp, t.SeedSource}
- if _, err = tx.ExecContext(ctx, `INSERT INTO routes (`+routeCols+`) VALUES (`+strings.TrimSuffix(strings.Repeat("?,", len(values)), ",")+`)`, values...); err != nil {
+ if _, err = tx.ExecContext(ctx, `INSERT INTO routes (`+routeCols+`) VALUES (`+placeholders(routeColCount)+`)`, routeArgs(next.Route)...); err != nil {
return err
}
for _, t := range next.Timers {
diff --git a/internal/registry/importjournal.go b/internal/registry/importjournal.go
new file mode 100644
index 000000000..446f0130b
--- /dev/null
+++ b/internal/registry/importjournal.go
@@ -0,0 +1,262 @@
+package registry
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "reflect"
+)
+
+// The import journal (seeded-instances spec §5.7): one row per host this
+// box has swapped for an import, holding the complete original route, so
+// the box always knows what it replaced, whatever happened to the record.
+const (
+ JournalSwapped = "swapped"
+ JournalRollingBack = "rolling-back"
+ JournalCleanupPending = "cleanup-pending"
+)
+
+// ImportGateClosed is the import gate a rollback installs: it matches no
+// token's hash (seeds.ClosedGate holds the same value).
+const ImportGateClosed = "closed"
+
+// ErrImportJournal is a journal transaction whose precondition did not
+// hold: no row or another seed's, a row already there, a changed route.
+var ErrImportJournal = errors.New("registry: the import journal does not allow this")
+
+// ImportJournal is one import_journal row. Original is the route the swap
+// replaced, exactly as it was stored.
+type ImportJournal struct {
+ Host string
+ SeedID string
+ Original Route
+ RecordKey string
+ State string
+ FenceGen uint64
+ At int64
+}
+
+func (r *Registry) migrateImportJournal() error {
+ _, err := r.sql.Exec(`CREATE TABLE IF NOT EXISTS import_journal (
+ host TEXT PRIMARY KEY,
+ seed_id TEXT NOT NULL,
+ original TEXT NOT NULL,
+ record_key TEXT NOT NULL DEFAULT '',
+ state TEXT NOT NULL,
+ fence_gen INTEGER NOT NULL DEFAULT 0,
+ at INTEGER NOT NULL DEFAULT 0
+);`)
+ return err
+}
+
+type queryer interface {
+ QueryRow(query string, args ...any) *sql.Row
+}
+
+type txer interface {
+ queryer
+ Exec(query string, args ...any) (sql.Result, error)
+}
+
+const journalCols = `host, seed_id, original, record_key, state, fence_gen, at`
+
+func scanJournalRow(scan func(...any) error) (ImportJournal, error) {
+ var j ImportJournal
+ var orig string
+ if err := scan(&j.Host, &j.SeedID, &orig, &j.RecordKey, &j.State, &j.FenceGen, &j.At); err != nil {
+ return j, err
+ }
+ if err := json.Unmarshal([]byte(orig), &j.Original); err != nil {
+ return j, fmt.Errorf("registry: import journal %s: original row: %w", j.Host, err)
+ }
+ return j, nil
+}
+
+func scanJournal(q queryer, host string) (*ImportJournal, error) {
+ j, err := scanJournalRow(q.QueryRow(`SELECT `+journalCols+` FROM import_journal WHERE host = ?`, host).Scan)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &j, nil
+}
+
+// ImportJournal returns host's journal row, or nil when there is none.
+func (r *Registry) ImportJournal(host string) (*ImportJournal, error) {
+ return scanJournal(r.sql, host)
+}
+
+// ImportJournals returns every journal row, ordered by host.
+func (r *Registry) ImportJournals() ([]ImportJournal, error) {
+ rows, err := r.sql.Query(`SELECT ` + journalCols + ` FROM import_journal ORDER BY host`)
+ if err != nil {
+ return nil, err
+ }
+ defer rows.Close()
+ var out []ImportJournal
+ for rows.Next() {
+ j, err := scanJournalRow(rows.Scan)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, j)
+ }
+ return out, rows.Err()
+}
+
+// inTx runs fn in one BEGIN IMMEDIATE transaction at synchronous=FULL (the
+// claim transactions' form). IMMEDIATE takes the write lock before the first
+// read, so another process's write waits on busy_timeout instead of failing
+// the upgrade; FULL makes the commit durable before the caller acts on it
+// (the swap's record CAS follows it). The registry's pool holds one
+// connection (Open), so in-process writers serialise on it.
+func (r *Registry) inTx(fn func(tx txer) error) error {
+ tx, err := r.beginClaimTransaction(context.Background(), true)
+ if err != nil {
+ return err
+ }
+ defer tx.Rollback()
+ if err := fn(tx); err != nil {
+ return err
+ }
+ return tx.Commit()
+}
+
+func getRouteTx(tx txer, host string) (*Route, error) {
+ t, err := scanRoute(tx.QueryRow(`SELECT `+routeCols+` FROM routes WHERE host = ?`, host).Scan)
+ if errors.Is(err, sql.ErrNoRows) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return &t, nil
+}
+
+// insertRouteTx inserts t exactly as given (no Upsert defaults).
+func insertRouteTx(tx txer, t Route) error {
+ _, err := tx.Exec(`INSERT INTO routes (`+routeCols+`) VALUES (`+placeholders(routeColCount)+`)`, routeArgs(t)...)
+ return err
+}
+
+// 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.
+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 == "" {
+ return fmt.Errorf("%w: the swap for %s is not well formed", ErrImportJournal, j.Host)
+ }
+ orig, err := json.Marshal(j.Original)
+ if err != nil {
+ return err
+ }
+ return r.inTx(func(tx txer) error {
+ if cur, err := scanJournal(tx, j.Host); err != nil {
+ return err
+ } else if cur != nil {
+ return fmt.Errorf("%w: %s already has a journal row for seed %s", ErrImportJournal, j.Host, cur.SeedID)
+ }
+ stored, err := getRouteTx(tx, j.Host)
+ if err != nil {
+ return err
+ }
+ if stored == nil || !reflect.DeepEqual(*stored, j.Original) {
+ return fmt.Errorf("%w: the route at %s changed since it was checked", ErrImportJournal, j.Host)
+ }
+ if _, err := tx.Exec(`DELETE FROM routes WHERE host = ?`, j.Host); err != nil {
+ return err
+ }
+ if err := insertRouteTx(tx, normalizeRoute(next)); err != nil {
+ return err
+ }
+ _, err = tx.Exec(`INSERT INTO import_journal (`+journalCols+`) VALUES (?,?,?,?,?,0,?)`,
+ j.Host, j.SeedID, string(orig), j.RecordKey, JournalSwapped, j.At)
+ return err
+ })
+}
+
+// journalFor is host's journal row, which must name seedID and be in one of
+// states.
+func journalFor(tx txer, host, seedID string, states ...string) (*ImportJournal, error) {
+ j, err := scanJournal(tx, host)
+ if err != nil {
+ return nil, err
+ }
+ if j == nil || j.SeedID != seedID {
+ return nil, fmt.Errorf("%w: no journal row for %s seed %s", ErrImportJournal, host, seedID)
+ }
+ for _, s := range states {
+ if j.State == s {
+ return j, nil
+ }
+ }
+ return nil, fmt.Errorf("%w: %s's journal is %s", ErrImportJournal, host, j.State)
+}
+
+// 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.
+func (r *Registry) CloseImportGate(host, seedID string, fenceGen uint64) error {
+ return r.inTx(func(tx txer) error {
+ if _, err := journalFor(tx, host, seedID, JournalSwapped, JournalRollingBack); err != nil {
+ return err
+ }
+ if _, err := tx.Exec(`UPDATE import_journal SET state = ?, fence_gen = ? WHERE host = ?`, JournalRollingBack, fenceGen, host); err != nil {
+ return err
+ }
+ _, err := tx.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, host)
+ return err
+ })
+}
+
+// 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.
+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 {
+ 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
+ })
+}
+
+// CommitImport clears the gate and retires the journal in one transaction
+// (spec §5.4 "Commit").
+func (r *Registry) CommitImport(host, seedID string) error {
+ return r.inTx(func(tx txer) error {
+ if _, err := journalFor(tx, host, seedID, JournalSwapped); err != nil {
+ return err
+ }
+ if _, err := tx.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, host); err != nil {
+ return err
+ }
+ _, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host)
+ return err
+ })
+}
+
+// DeleteImportJournal is the rollback's step 7.
+func (r *Registry) DeleteImportJournal(host, seedID string) error {
+ return r.inTx(func(tx txer) error {
+ if _, err := journalFor(tx, host, seedID, JournalCleanupPending); err != nil {
+ return err
+ }
+ _, err := tx.Exec(`DELETE FROM import_journal WHERE host = ?`, host)
+ return err
+ })
+}
diff --git a/internal/registry/importjournal_test.go b/internal/registry/importjournal_test.go
new file mode 100644
index 000000000..0ef0d9206
--- /dev/null
+++ b/internal/registry/importjournal_test.go
@@ -0,0 +1,340 @@
+package registry
+
+import (
+ "context"
+ "errors"
+ "path/filepath"
+ "reflect"
+ "testing"
+)
+
+// 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"
+)
+
+func legacyRow() Route {
+ return Route{Host: importHost, Kind: "service", Addr: "127.0.0.1:8401", Status: "ready",
+ App: "shop", Account: "bab", Channel: "stable", Unit: "shop@old.service", Caps: "", CreatedAt: 1700000000}
+}
+
+func execRow() Route {
+ return Route{Host: importHost, Kind: "instance", Socket: "/run/carlos/instances/old/app.sock",
+ DBPath: "/data/instances/old/old.shop.example.com.db", Status: "ready", App: "shop", Account: "bbw",
+ Channel: "stable", Hibernate: true, Provisioned: true, ImportGate: "abc123", CreatedAt: 1700000001}
+}
+
+func TestSwapCommitAndRestore(t *testing.T) {
+ reg := openTestRegistry(t)
+ old := legacyRow()
+ if err := reg.Upsert(old); err != nil {
+ t.Fatal(err)
+ }
+ stored, _ := reg.Get(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 {
+ t.Fatal(err)
+ }
+ got, _ := reg.Get(old.Host)
+ if got.Kind != "instance" || got.ImportGate != "abc123" || got.Account != "bbw" {
+ t.Fatalf("after swap: %+v", got)
+ }
+ jr, _ := reg.ImportJournal(old.Host)
+ if jr == nil || jr.State != JournalSwapped || !reflect.DeepEqual(jr.Original, *stored) || jr.RecordKey != j.RecordKey || jr.At != 1 {
+ 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) {
+ t.Fatalf("second swap: %v", err)
+ }
+ // Close the gate, restore the route exactly, and the journal says where it is.
+ if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := reg.Get(old.Host); got.ImportGate != ImportGateClosed {
+ t.Fatalf("gate = %q, want closed", got.ImportGate)
+ }
+ // Commit is only from swapped: a rolling-back import cannot commit.
+ if err := reg.CommitImport(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("commit while rolling back: %v", err)
+ }
+ // Closing again resumes (idempotent) and keeps the latest generation.
+ if err := reg.CloseImportGate(old.Host, j.SeedID, 77); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.RestoreImportRoute(old.Host, j.SeedID); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := reg.Get(old.Host); !reflect.DeepEqual(*got, *stored) {
+ t.Fatalf("restored %+v, want %+v", got, stored)
+ }
+ if jr, _ := reg.ImportJournal(old.Host); jr == nil || jr.State != JournalCleanupPending || jr.FenceGen != 77 {
+ t.Fatalf("journal after restore = %+v", jr)
+ }
+ // cleanup-pending never goes back: no second restore, no gate close.
+ if err := reg.RestoreImportRoute(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("restore from cleanup-pending: %v", err)
+ }
+ if err := reg.CloseImportGate(old.Host, j.SeedID, 78); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("close from cleanup-pending: %v", err)
+ }
+ if err := reg.DeleteImportJournal(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("delete with another id: %v", err)
+ }
+ if err := reg.DeleteImportJournal(old.Host, j.SeedID); err != nil {
+ t.Fatal(err)
+ }
+ if jr, _ := reg.ImportJournal(old.Host); jr != nil {
+ t.Fatalf("journal after delete = %+v", jr)
+ }
+ if err := reg.DeleteImportJournal(old.Host, j.SeedID); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("delete with no row: %v", err)
+ }
+}
+
+func TestCommitImportClearsTheGateAndTheJournal(t *testing.T) {
+ reg := openTestRegistry(t)
+ old := legacyRow()
+ if err := reg.Upsert(old); err != nil {
+ t.Fatal(err)
+ }
+ stored, _ := reg.Get(old.Host)
+ j := ImportJournal{Host: old.Host, SeedID: importSeedID, Original: *stored}
+ if err := reg.SwapForImport(j, execRow()); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.CommitImport(old.Host, "sd_bbbbbbbbbbbbbbbbbbbbbbbbbb"); !errors.Is(err, ErrImportJournal) {
+ t.Fatalf("commit with another id: %v", err)
+ }
+ before, _ := reg.Fingerprint()
+ if err := reg.CommitImport(old.Host, j.SeedID); err != nil {
+ t.Fatal(err)
+ }
+ got, _ := reg.Get(old.Host)
+ jr, _ := reg.ImportJournal(old.Host)
+ after, _ := reg.Fingerprint()
+ if got.ImportGate != "" || jr != nil || before == after {
+ 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.ImportGate = ""
+ if !reflect.DeepEqual(*got, want) {
+ t.Fatalf("committed row %+v, want %+v", *got, want)
+ }
+}
+
+func TestSwapForImportRefusesAChangedRow(t *testing.T) {
+ reg := openTestRegistry(t)
+ old := legacyRow()
+ if err := reg.Upsert(old); err != nil {
+ t.Fatal(err)
+ }
+ stored, _ := reg.Get(old.Host)
+ changed := *stored
+ changed.Addr = "127.0.0.1:9999"
+ 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) {
+ t.Fatalf("swap over a changed row: %v", err)
+ }
+ if got, _ := reg.Get(old.Host); got.Addr != "127.0.0.1:9999" {
+ t.Fatal("the changed row was replaced")
+ }
+ if jr, _ := reg.ImportJournal(old.Host); jr != nil {
+ t.Fatalf("a refused swap left a journal row: %+v", jr)
+ }
+}
+
+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) {
+ t.Fatalf("swap with no stored row: %v", err)
+ }
+ if err := reg.Upsert(old); err != nil {
+ t.Fatal(err)
+ }
+ stored, _ := reg.Get(old.Host)
+ j.Original = *stored
+ ungated := execRow()
+ ungated.ImportGate = ""
+ other := execRow()
+ other.Host = "new.shop.example.com"
+ noID := j
+ noID.SeedID = ""
+ 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()},
+ } {
+ if err := reg.SwapForImport(c.j, c.next); !errors.Is(err, ErrImportJournal) {
+ t.Errorf("%s: %v", name, err)
+ }
+ }
+ if got, _ := reg.Get(old.Host); !reflect.DeepEqual(*got, *stored) {
+ t.Fatalf("a refused swap changed the row: %+v", got)
+ }
+ if js, err := reg.ImportJournals(); err != nil || len(js) != 0 {
+ t.Fatalf("journals = %+v, %v", js, err)
+ }
+}
+
+func TestUpsertNeverChangesTheImportGate(t *testing.T) {
+ reg := openTestRegistry(t)
+ row := execRow()
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ row.ImportGate = ""
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := reg.Get(row.Host); got.ImportGate != "abc123" {
+ t.Fatalf("an Upsert cleared the gate: %q", got.ImportGate)
+ }
+ row.ImportGate = "def456"
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ if got, _ := reg.Get(row.Host); got.ImportGate != "abc123" {
+ t.Fatalf("an Upsert changed the gate: %q", got.ImportGate)
+ }
+}
+
+func TestImportJournalsListsEveryRow(t *testing.T) {
+ reg := openTestRegistry(t)
+ for i, host := range []string{"b.shop.example.com", "a.shop.example.com"} {
+ old := legacyRow()
+ old.Host = host
+ if err := reg.Upsert(old); err != nil {
+ t.Fatal(err)
+ }
+ stored, _ := reg.Get(host)
+ next := execRow()
+ next.Host = host
+ j := ImportJournal{Host: host, SeedID: importSeedID, Original: *stored, At: int64(i + 1)}
+ if err := reg.SwapForImport(j, next); err != nil {
+ t.Fatal(err)
+ }
+ }
+ js, err := reg.ImportJournals()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(js) != 2 || js[0].Host != "a.shop.example.com" || js[1].Host != "b.shop.example.com" ||
+ js[0].Original.Host != "a.shop.example.com" || js[0].At != 2 || js[1].State != JournalSwapped {
+ t.Fatalf("journals = %+v", js)
+ }
+}
+
+// The gate moves both digests, and an ungated registry digests as it did
+// before the column existed (the goldens in hold_test.go cover that half).
+func TestImportGateMovesBothFingerprints(t *testing.T) {
+ reg := openTestRegistry(t)
+ if err := reg.Upsert(legacyRow()); err != nil {
+ t.Fatal(err)
+ }
+ fp1, _ := reg.Fingerprint()
+ sfp1, _ := reg.SnapshotFingerprint()
+ if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = 'abc123' WHERE host = ?`, importHost); err != nil {
+ t.Fatal(err)
+ }
+ fp2, _ := reg.Fingerprint()
+ sfp2, _ := reg.SnapshotFingerprint()
+ if fp1 == fp2 || sfp1 == sfp2 {
+ t.Fatal("setting the gate did not move both fingerprints")
+ }
+ if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = ? WHERE host = ?`, ImportGateClosed, importHost); err != nil {
+ t.Fatal(err)
+ }
+ if fp3, _ := reg.Fingerprint(); fp3 == fp2 {
+ t.Fatal("closing the gate did not move the edge's fingerprint")
+ }
+ if _, err := reg.sql.Exec(`UPDATE routes SET import_gate = '' WHERE host = ?`, importHost); err != nil {
+ t.Fatal(err)
+ }
+ fp4, _ := reg.Fingerprint()
+ sfp4, _ := reg.SnapshotFingerprint()
+ if fp4 != fp1 || sfp4 != sfp1 {
+ t.Fatal("clearing the gate did not restore the ungated digests")
+ }
+}
+
+// A registry from before the column gains it on open, every time.
+func TestImportGateMigrationOnALegacyDB(t *testing.T) {
+ path := filepath.Join(t.TempDir(), "registry.db")
+ r, err := Open(path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := r.Upsert(legacyRow()); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`ALTER TABLE routes DROP COLUMN import_gate`); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := r.sql.Exec(`DROP TABLE import_journal`); err != nil {
+ t.Fatal(err)
+ }
+ r.Close()
+ for i := 0; i < 2; i++ {
+ r, err = Open(path)
+ if err != nil {
+ t.Fatalf("open %d: %v", i, err)
+ }
+ rt, err := r.Get(importHost)
+ if err != nil || rt == nil || rt.ImportGate != "" {
+ t.Fatalf("open %d: %+v %v", i, rt, err)
+ }
+ if js, err := r.ImportJournals(); err != nil || len(js) != 0 {
+ t.Fatalf("open %d: journals %+v %v", i, js, err)
+ }
+ r.Close()
+ }
+}
+
+// Preflight M6: the cold handover's own INSERT carries import_gate, so a
+// restored row keeps its gate.
+func TestRestoreClaimRowsCarriesTheImportGate(t *testing.T) {
+ ctx := context.Background()
+ reg := openTestRegistry(t)
+ owner := ClaimOwnership{"inc", 7, "installation-7", "route-7"}
+ row := execRow()
+ row.ClaimOwnership = owner
+ if err := reg.Upsert(row); err != nil {
+ t.Fatal(err)
+ }
+ old, err := reg.Get(row.Host)
+ if err != nil || old == nil || old.ImportGate != "abc123" {
+ t.Fatalf("stored = %+v, %v", old, err)
+ }
+ targets := []ClaimRow{{Kind: "route", Host: row.Host, Ownership: owner}}
+ if err := reg.RetireClaimRows(ctx, "handover-old", targets, func([]ClaimRowArchive) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ next := *old
+ next.ClaimOwnership = ClaimOwnership{"inc", 8, "installation-8", "route-8"}
+ if err := reg.RestoreClaimRows(ctx, "handover", "handover-old", ColdRows{Route: *old}, ColdRows{Route: next}); err != nil {
+ t.Fatal(err)
+ }
+ got, err := reg.Get(row.Host)
+ if err != nil || got == nil {
+ t.Fatalf("restored = %+v, %v", got, err)
+ }
+ if *got != next {
+ t.Fatalf("restored %+v, want %+v", *got, next)
+ }
+ // A replay of the same handover is a success, and reads the gate back.
+ if err := reg.RestoreClaimRows(ctx, "handover", "handover-old", ColdRows{Route: *old}, ColdRows{Route: next}); err != nil {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/registry/registry.go b/internal/registry/registry.go
index 55c69d407..32345ef41 100644
--- a/internal/registry/registry.go
+++ b/internal/registry/registry.go
@@ -75,6 +75,14 @@ type Route struct {
// it, never the record. Upsert keeps a non-empty value over an empty
// one; SetSeedSource converges it. Not in Fingerprint.
SeedSource string
+ // ImportGate is an uncommitted import's gate (seeded-instances spec
+ // §5.4): the sha256 of its verify token, or ImportGateClosed during a
+ // rollback. Set: the edge answers 503 to every request without the
+ // matching token, and ticks skip the row. Only the import transactions
+ // (importjournal.go) write it; Upsert's conflict path never changes it.
+ // In Fingerprint (the edge must reload on a change), folded only when
+ // non-empty.
+ ImportGate string
LastError string
DBPath string
@@ -529,6 +537,7 @@ CREATE TABLE IF NOT EXISTS routes (
`ALTER TABLE routes ADD COLUMN replica INTEGER NOT NULL DEFAULT 0`,
`ALTER TABLE routes ADD COLUMN hold_op TEXT NOT NULL DEFAULT ''`,
`ALTER TABLE routes ADD COLUMN seed_source TEXT NOT NULL DEFAULT ''`,
+ `ALTER TABLE routes ADD COLUMN import_gate TEXT NOT NULL DEFAULT ''`,
} {
if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
@@ -611,6 +620,9 @@ CREATE INDEX IF NOT EXISTS timers_due ON timers (next_at) WHERE next_at > 0;`);
if _, err := r.sql.Exec(`ALTER TABLE timers ADD COLUMN registration TEXT NOT NULL DEFAULT ''`); err != nil && !strings.Contains(err.Error(), "duplicate column") {
return err
}
+ if err := r.migrateImportJournal(); err != nil {
+ return err
+ }
if err := r.migrateAccounts(); err != nil {
return err
}
@@ -774,18 +786,26 @@ func (r *Registry) Accounts() ([]Account, error) {
return out, rows.Err()
}
-// routeCols is the column list every route read and the one route write
-// share — 48 names, in this exact order. scanRoute's destinations, Upsert's
-// placeholder row and Upsert's argument list all count from here: adding a
-// column means adding it in FOUR places, and a miscount is a silent column
-// shift rather than an error (TestUpsertRoundTripsAllColumns is what catches
-// one).
+// routeCols is the column list every route read and every route insert
+// share — routeColCount names, in this exact order. scanRoute's
+// destinations and routeArgs's values count from here: adding a column means
+// adding it to this list, routeColCount, scanRoute and routeArgs, and a
+// miscount is a silent column shift rather than an error
+// (TestUpsertRoundTripsAllColumns is what catches one).
const routeCols = `host, slug, kind, socket, addr, status, last_error, db_path,
version, prev_version, app, pool, region, channel, hibernate, canary,
replication, next_due, backing, status_named, platform, dns_status,
dns_checked_at, cert_expires_at, account, environment, caps, tunnel,
provisioned, seed, unit, fleet_box, addrs, health, steering, gate, created_at,
-args, idle_seconds, max_run_seconds, replicas, replica, claim_incarnation, claim_stamp, claim_owner, claim_row_id, hold_op, seed_source`
+args, idle_seconds, max_run_seconds, replicas, replica, claim_incarnation, claim_stamp, claim_owner, claim_row_id, hold_op, seed_source, import_gate`
+
+// routeColCount is the number of names in routeCols.
+const routeColCount = 49
+
+// placeholders is n comma-joined bind parameters.
+func placeholders(n int) string {
+ return strings.TrimSuffix(strings.Repeat("?,", n), ",")
+}
func scanRoute(scan func(...any) error) (Route, error) {
var t Route
@@ -796,7 +816,7 @@ func scanRoute(scan func(...any) error) (Route, error) {
&sn, &plat, &t.DNSStatus, &t.DNSCheckedAt, &t.CertExpiresAt, &t.Account,
&t.Environment, &t.Caps, &t.Tunnel, &prov, &t.Seed, &t.Unit, &t.FleetBox,
&t.Addrs, &t.Health, &t.Steering, &t.Gate, &t.CreatedAt,
- &t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID, &t.HoldOp, &t.SeedSource)
+ &t.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID, &t.HoldOp, &t.SeedSource, &t.ImportGate)
t.Hibernate = hib != 0
t.Canary = can != 0
t.StatusNamed = sn != 0
@@ -982,42 +1002,9 @@ func (r *Registry) Delete(host string) error {
// Route53 latency records — a real incident this repo caught in review before
// it shipped, not a hypothetical.
func (r *Registry) Upsert(t Route) error {
- if t.CreatedAt == 0 {
- t.CreatedAt = time.Now().Unix()
- }
- if t.Status == "" {
- t.Status = "ready"
- }
- if t.Kind == "" {
- t.Kind = "instance"
- }
- // The replication invariant, INSERT half (the ON CONFLICT half is the
- // `excluded.hibernate = 0` arm below): a non-empty class on a row that
- // does not hibernate is a column that lies. Nothing reads it there —
- // the activator's wake path is its only reader, and box-wide
- // /etc/litestream.yml covers every non-hibernating database at its own
- // interval — so a stored `daily` would be reported by `carlos routes`
- // while the database was streamed every second. See SetHibernate.
- if !t.Hibernate {
- t.Replication = ""
- }
- // #162's replica invariant, its clean form: a replica SET on a row that
- // IS itself a replica is a column that lies — a replica of a replica is
- // not a shape any supported writer produces, and a replica row's
- // placement is its own Region. Dropped silently, like the replication
- // rung above.
- if t.Replica {
- t.Replicas = ""
- }
- b2i := func(b bool) int {
- if b {
- return 1
- }
- return 0
- }
res, err := r.sql.Exec(`
INSERT INTO routes (`+routeCols+`)
-VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+VALUES (`+placeholders(routeColCount)+`)
ON CONFLICT(host) DO UPDATE SET
slug=excluded.slug, kind=excluded.kind, socket=excluded.socket,
addr=excluded.addr,
@@ -1049,14 +1036,7 @@ ON CONFLICT(host) DO UPDATE SET
seed_source=CASE WHEN excluded.seed_source = '' THEN routes.seed_source ELSE excluded.seed_source END
WHERE routes.claim_incarnation=excluded.claim_incarnation AND routes.claim_stamp=excluded.claim_stamp
AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.claim_row_id`,
- t.Host, t.Slug, t.Kind, t.Socket, t.Addr, t.Status, t.LastError,
- t.DBPath, t.Version, t.PrevVersion, t.App, t.Pool, t.Region, t.Channel,
- b2i(t.Hibernate), b2i(t.Canary), t.Replication, t.NextDue, t.Backing,
- b2i(t.StatusNamed), b2i(t.Platform), t.DNSStatus, t.DNSCheckedAt,
- t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel,
- b2i(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health,
- t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds,
- t.Replicas, b2i(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp, t.SeedSource)
+ routeArgs(normalizeRoute(t))...)
if err != nil {
return err
}
@@ -1067,6 +1047,56 @@ AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.cla
return err
}
+// normalizeRoute applies Upsert's defaults and invariants to a row about to
+// be inserted. The import swap's new row goes through it too; a restored
+// original does not (it goes back exactly as it was stored).
+func normalizeRoute(t Route) Route {
+ if t.CreatedAt == 0 {
+ t.CreatedAt = time.Now().Unix()
+ }
+ if t.Status == "" {
+ t.Status = "ready"
+ }
+ if t.Kind == "" {
+ t.Kind = "instance"
+ }
+ // The replication invariant, INSERT half (the ON CONFLICT half is the
+ // `excluded.hibernate = 0` arm in Upsert): a non-empty class on a row that
+ // does not hibernate is a column that lies. Nothing reads it there —
+ // the activator's wake path is its only reader, and box-wide
+ // /etc/litestream.yml covers every non-hibernating database at its own
+ // interval — so a stored `daily` would be reported by `carlos routes`
+ // while the database was streamed every second. See SetHibernate.
+ if !t.Hibernate {
+ t.Replication = ""
+ }
+ // #162's replica invariant, its clean form: a replica SET on a row that
+ // IS itself a replica is a column that lies — a replica of a replica is
+ // not a shape any supported writer produces, and a replica row's
+ // placement is its own Region. Dropped silently, like the replication
+ // rung above.
+ if t.Replica {
+ t.Replicas = ""
+ }
+ return t
+}
+
+// routeArgs is t's values in routeCols order, exactly as given: every route
+// insert binds through it, so a column added later is added here once.
+func routeArgs(t Route) []any {
+ return []any{
+ t.Host, t.Slug, t.Kind, t.Socket, t.Addr, t.Status, t.LastError,
+ t.DBPath, t.Version, t.PrevVersion, t.App, t.Pool, t.Region, t.Channel,
+ b2iDB(t.Hibernate), b2iDB(t.Canary), t.Replication, t.NextDue, t.Backing,
+ b2iDB(t.StatusNamed), b2iDB(t.Platform), t.DNSStatus, t.DNSCheckedAt,
+ t.CertExpiresAt, t.Account, t.Environment, t.Caps, t.Tunnel,
+ b2iDB(t.Provisioned), t.Seed, t.Unit, t.FleetBox, t.Addrs, t.Health,
+ t.Steering, t.Gate, t.CreatedAt, t.Args, t.IdleSeconds, t.MaxRunSeconds,
+ t.Replicas, b2iDB(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp, t.SeedSource,
+ t.ImportGate,
+ }
+}
+
// SetSeedSource converges a row's seed column from its record.
func (r *Registry) SetSeedSource(host, v string) error {
res, err := r.sql.Exec(`UPDATE routes SET seed_source = ? WHERE host = ?`, v, host)
@@ -2459,7 +2489,7 @@ func (r *Registry) Fingerprint() (string, error) {
}
rows, err := db.Query(`
SELECT rowid, status, socket, addr, version, hibernate, channel, backing, tunnel,
- db_path, region, fleet_box, addrs, health, gate, hold_op,
+ db_path, region, fleet_box, addrs, health, gate, hold_op, import_gate,
claim_incarnation, claim_stamp, claim_owner, claim_row_id
FROM routes ORDER BY rowid`)
if err != nil {
@@ -2471,9 +2501,9 @@ FROM routes ORDER BY rowid`)
for rows.Next() {
var rowid, hibernate int64
var claim ClaimOwnership
- var status, socket, addr, version, channel, backing, tunnel, dbPath, region, fleetBox, addrs, health, gate, holdOp string
+ var status, socket, addr, version, channel, backing, tunnel, dbPath, region, fleetBox, addrs, health, gate, holdOp, importGate string
if err := rows.Scan(&rowid, &status, &socket, &addr, &version, &hibernate,
- &channel, &backing, &tunnel, &dbPath, &region, &fleetBox, &addrs, &health, &gate, &holdOp,
+ &channel, &backing, &tunnel, &dbPath, &region, &fleetBox, &addrs, &health, &gate, &holdOp, &importGate,
&claim.ClaimIncarnation, &claim.ClaimStamp, &claim.ClaimOwner, &claim.ClaimRowID); err != nil {
return "", err
}
@@ -2487,6 +2517,12 @@ FROM routes ORDER BY rowid`)
if holdOp != "" {
fmt.Fprintf(h, "\x1fhold_op=%s", holdOp)
}
+ // An import's gate follows the same rule: the edge must reload when a
+ // gate is set, closed or cleared, and an ungated registry digests as
+ // it did before the column existed.
+ if importGate != "" {
+ fmt.Fprintf(h, "\x1fimport_gate=%q", importGate)
+ }
// Ownership follows the same rule: the all-default value is omitted,
// and each non-default field is tagged and quoted so a partial value
// cannot alias another field's.
@@ -2728,6 +2764,14 @@ func (r *Registry) SnapshotFingerprint() (string, error) {
}
continue
}
+ // So does import_gate: a registry with no gated rows digests
+ // exactly as it did before the column existed.
+ if cols[i] == "import_gate" {
+ if v.Valid && v.String != "" {
+ fmt.Fprintf(h, "import_gate=%s\x1f", v.String)
+ }
+ continue
+ }
// Claim ownership columns follow the hold_op rule: default
// values are omitted, anything else is tagged and quoted.
if snapshotClaimCols[cols[i]] {
diff --git a/internal/registry/registry_test.go b/internal/registry/registry_test.go
index 3456c00a2..501d3a2f7 100644
--- a/internal/registry/registry_test.go
+++ b/internal/registry/registry_test.go
@@ -82,6 +82,7 @@ func TestUpsertRoundTripsAllColumns(t *testing.T) {
Gate: "$2a$10$distinct",
CreatedAt: 1700000000,
SeedSource: `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`,
+ ImportGate: "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef",
}
if err := reg.Upsert(want); err != nil {
t.Fatal(err)