carlos / platform Public

registry: the seed_source column (Plan 4 Task 1)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago 0c9b6703b87f3ea558f29d20ff711db55ceef988 parent 7cb5fc2

4 files changed, +104 −11

internal/registry/coldhandover.go +1 −1
internal/registry/registry.go +38 −6
internal/registry/registry_test.go +5 −4
internal/registry/seedsource_test.go +60 −0

diff --git a/internal/registry/coldhandover.go b/internal/registry/coldhandover.go

index 6e832076a..980afd118 100644

--- a/internal/registry/coldhandover.go

+++ b/internal/registry/coldhandover.go

@@ -102,7 +102,7 @@ func (r *Registry) RestoreClaimRows(ctx context.Context, id, retirement string,

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}

+ 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 {

return err

}

diff --git a/internal/registry/registry.go b/internal/registry/registry.go

index db75d890e..55c69d407 100644

--- a/internal/registry/registry.go

+++ b/internal/registry/registry.go

@@ -69,6 +69,12 @@ type Route struct {

// folded in only when non-empty so an unheld registry hashes exactly as it

// did before the column existed.

HoldOp string

+ // SeedSource is the record's seed as this row needs it

+ // (seeds.RowSeed's JSON, "" for none), written by the provision pass

+ // (seeded-instances spec §3.4): the activator's restore fallback reads

+ // it, never the record. Upsert keeps a non-empty value over an empty

+ // one; SetSeedSource converges it. Not in Fingerprint.

+ SeedSource string

LastError string

DBPath string

@@ -522,6 +528,7 @@ CREATE TABLE IF NOT EXISTS routes (

`ALTER TABLE routes ADD COLUMN replicas TEXT NOT NULL DEFAULT ''`,

`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 ''`,

} {

if _, err := r.sql.Exec(stmt); err != nil && !strings.Contains(err.Error(), "duplicate column") {

return err

@@ -768,7 +775,7 @@ func (r *Registry) Accounts() ([]Account, error) {

}

// routeCols is the column list every route read and the one route write

-// share — 47 names, in this exact order. scanRoute's destinations, Upsert's

+// 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

@@ -778,7 +785,7 @@ 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`

+args, idle_seconds, max_run_seconds, replicas, replica, claim_incarnation, claim_stamp, claim_owner, claim_row_id, hold_op, seed_source`

func scanRoute(scan func(...any) error) (Route, error) {

var t Route

@@ -789,7 +796,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.Args, &t.IdleSeconds, &t.MaxRunSeconds, &t.Replicas, &rep, &t.ClaimIncarnation, &t.ClaimStamp, &t.ClaimOwner, &t.ClaimRowID, &t.HoldOp, &t.SeedSource)

t.Hibernate = hib != 0

t.Canary = can != 0

t.StatusNamed = sn != 0

@@ -1010,7 +1017,7 @@ func (r *Registry) Upsert(t Route) error {

}

res, err := r.sql.Exec(`

INSERT INTO routes (`+routeCols+`)

-VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)

+VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)

ON CONFLICT(host) DO UPDATE SET

slug=excluded.slug, kind=excluded.kind, socket=excluded.socket,

addr=excluded.addr,

@@ -1038,7 +1045,8 @@ ON CONFLICT(host) DO UPDATE SET

gate=CASE WHEN excluded.gate = '' THEN routes.gate ELSE excluded.gate END,

args=CASE WHEN excluded.args = '' THEN routes.args ELSE excluded.args END,

idle_seconds=CASE WHEN excluded.idle_seconds = 0 THEN routes.idle_seconds ELSE excluded.idle_seconds END,

- max_run_seconds=CASE WHEN excluded.max_run_seconds = 0 THEN routes.max_run_seconds ELSE excluded.max_run_seconds END

+ max_run_seconds=CASE WHEN excluded.max_run_seconds = 0 THEN routes.max_run_seconds ELSE excluded.max_run_seconds END,

+ 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,

@@ -1048,7 +1056,7 @@ AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.cla

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.Replicas, b2i(t.Replica), t.ClaimIncarnation, t.ClaimStamp, t.ClaimOwner, t.ClaimRowID, t.HoldOp, t.SeedSource)

if err != nil {

return err

}

@@ -1059,6 +1067,22 @@ AND routes.claim_owner=excluded.claim_owner AND routes.claim_row_id=excluded.cla

return err

}

+// 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)

+ if err != nil {

+ return err

+ }

+ n, err := res.RowsAffected()

+ if err != nil {

+ return err

+ }

+ if n == 0 {

+ return fmt.Errorf("no route for host %q", host)

+ }

+ return nil

// HoldStatuses are the row statuses the edge never serves, ticks refuse and

// no ordinary wake reaches (moves spec §5.2).

var HoldStatuses = map[string]bool{"moving": true, "landing": true, "restoring": true, "releasing": true}

@@ -2696,6 +2720,14 @@ func (r *Registry) SnapshotFingerprint() (string, error) {

}

continue

}

+ // seed_source follows the same rule: a registry with no seeded

+ // rows digests exactly as it did before the column existed.

+ if cols[i] == "seed_source" {

+ if v.Valid && v.String != "" {

+ fmt.Fprintf(h, "seed_source=%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 b479ca6e1..3456c00a2 100644

--- a/internal/registry/registry_test.go

+++ b/internal/registry/registry_test.go

@@ -77,10 +77,11 @@ func TestUpsertRoundTripsAllColumns(t *testing.T) {

Environment: "demo", Caps: "host-role,port:25",

Hibernate: true, Canary: true, Replication: "batched",

NextDue: 4200, Backing: BackingUnit, Unit: "console.service",

- Steering: "latency",

- Replicas: "ie,uk,us",

- Gate: "$2a$10$distinct",

- CreatedAt: 1700000000,

+ Steering: "latency",

+ Replicas: "ie,uk,us",

+ Gate: "$2a$10$distinct",

+ CreatedAt: 1700000000,

+ SeedSource: `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`,

}

if err := reg.Upsert(want); err != nil {

t.Fatal(err)

diff --git a/internal/registry/seedsource_test.go b/internal/registry/seedsource_test.go

new file mode 100644

index 000000000..04698da22

--- /dev/null

+++ b/internal/registry/seedsource_test.go

@@ -0,0 +1,60 @@

+package registry

+import "testing"

+func TestSeedSourceColumn(t *testing.T) {

+ reg := openTestRegistry(t)

+ const col = `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`

+ if err := reg.Upsert(Route{Host: "c.example.com", Kind: "instance", SeedSource: col}); err != nil {

+ t.Fatal(err)

+ }

+ // A re-add that knows nothing of seeds keeps the column.

+ if err := reg.Upsert(Route{Host: "c.example.com", Kind: "instance"}); err != nil {

+ t.Fatal(err)

+ }

+ if got, _ := reg.Get("c.example.com"); got.SeedSource != col {

+ t.Fatalf("after a seedless Upsert: %q", got.SeedSource)

+ }

+ placed := `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"placed"}`

+ if err := reg.SetSeedSource("c.example.com", placed); err != nil {

+ t.Fatal(err)

+ }

+ if got, _ := reg.Get("c.example.com"); got.SeedSource != placed {

+ t.Fatalf("after SetSeedSource: %q", got.SeedSource)

+ }

+ if err := reg.SetSeedSource("nope.example.com", placed); err == nil {

+ t.Fatal("SetSeedSource on no row: want an error")

+ }

+ before, _ := reg.Fingerprint()

+ _ = reg.SetSeedSource("c.example.com", col)

+ if after, _ := reg.Fingerprint(); after != before {

+ t.Fatal("the seed column moved the edge's fingerprint")

+ }

+// The snapshot digest sees the column when it is set and digests exactly as

+// before when it is not (the golden in TestFingerprintsUnchangedForUnheldRows

+// covers the unseeded half against the pre-column hash).

+func TestSeedSourceSnapshotFingerprint(t *testing.T) {

+ reg := openTestRegistry(t)

+ if err := reg.Upsert(Route{Host: "c.example.com", Kind: "instance", CreatedAt: 1700000000}); err != nil {

+ t.Fatal(err)

+ }

+ unseeded, err := reg.SnapshotFingerprint()

+ if err != nil {

+ t.Fatal(err)

+ }

+ if err := reg.SetSeedSource("c.example.com", `{"id":"sd_aaaaaaaaaaaaaaaaaaaaaaaaaa","kind":"replica","state":"ready"}`); err != nil {

+ t.Fatal(err)

+ }

+ seeded, _ := reg.SnapshotFingerprint()

+ if seeded == unseeded {

+ t.Fatal("SnapshotFingerprint did not move when seed_source was set")

+ }

+ if err := reg.SetSeedSource("c.example.com", ""); err != nil {

+ t.Fatal(err)

+ }

+ if back, _ := reg.SnapshotFingerprint(); back != unseeded {

+ t.Fatal("clearing seed_source did not restore the unseeded digest")

+ }