carlos / platform Public

#162: mint a replica route with its home's channel and replica_of

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev e5c88872b361d98cfc21be1f5aa312b5c8a36494 parent 4fb82f2
2 files changed, +204 −19
  • internal/provision/reconcile.go +80 −12
  • internal/provision/reconcile_test.go +124 −7
diff --git a/internal/provision/reconcile.go b/internal/provision/reconcile.go
index 7f4b6dd86..aa70ba875 100644
--- a/internal/provision/reconcile.go
+++ b/internal/provision/reconcile.go
@@ -75,6 +75,12 @@ type ReconcileConfig struct {
// path; the auto-purge that does still lives in domains, still refuses
// without a key.
ReleasePub ed25519.PublicKey
+ // OnRecords, when set, is called once per pass with every record this
+ // pass decoded — in scope or not. The siblings-file pass (spec §7)
+ // needs both sides of a home/replica pair and only one is ever in
+ // scope on a given box; this hands it the map the pass already built
+ // rather than making it list apps/ a second time.
+ OnRecords func(all map[string]Record)
}
// InboundAny and InboundNone are the closed vocabulary ReconcileConfig.Inbound
@@ -345,7 +351,11 @@ func (cfg ReconcileConfig) inScope(region, pool string, everywhere bool) bool {
// exclusion set from `r.Hibernate && r.DBPath != ""`, so a hibernating route
// whose row carried no path would be swept up by the dataDir glob and
// replicated box-wide as if it never slept.
-func routeFor(rec Record, dataDir, seed string) registry.Route {
+// home is the record rec.Home names, and is read ONLY when rec is a replica
+// (#162): a non-replica caller passes a zero Record and neither branch below
+// looks at it. A caller that could not find the home record must not reach
+// here at all — Reconcile protects such a replica rather than minting it.
+func routeFor(rec Record, home Record, dataDir, seed string) registry.Route {
rt := registry.Route{
Host: rec.Host, Slug: rec.Slug, Kind: rec.KindOrDefault(), Unit: rec.Unit,
App: rec.App, Account: rec.Account, Pool: rec.Pool, Region: rec.Region,
@@ -355,6 +365,13 @@ func routeFor(rec Record, dataDir, seed string) registry.Route {
Gate: rec.Gate,
Provisioned: true, Seed: seed, Status: "ready",
}
+ if rec.Role == "replica" {
+ // The link the domains pass and the edges-file writer read
+ // (spec §3), and the HOME's channel — one channel across the pair,
+ // and the replica record carries none of its own (spec §2).
+ rt.ReplicaOf = rec.Home
+ rt.Channel = home.Channel
+ }
if rec.DerivesPaths() {
rt.Socket = DefaultSocket(rec.Host)
rt.DBPath = DefaultDBPath(dataDir, rec.Host)
@@ -569,14 +586,24 @@ func shapeMismatch(existing, want registry.Route) string {
// one of the two callers is exactly the single-unguarded-write failure mode
// the pass exists to prevent (design §4.3). Guarding here means both call
// sites are safe by construction, not by each one remembering to check.
-func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, dryRun bool) {
- if rec.Channel != "" && existing.Channel != rec.Channel {
+//
+// home is routeFor's parameter with routeFor's contract: read only when rec
+// is a replica (#162), a zero Record otherwise.
+func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record, home Record, dryRun bool) {
+ // A replica follows its HOME's channel, not its own: one channel across
+ // the pair, so a `carlos canary` on the home carries to the replica on
+ // the next pass (spec §2).
+ channel := rec.Channel
+ if rec.Role == "replica" {
+ channel = home.Channel
+ }
+ if channel != "" && existing.Channel != channel {
if dryRun {
- log.Printf("provision: dry-run: would converge %s channel %s -> %s", rec.Host, existing.Channel, rec.Channel)
- } else if err := reg.SetChannel(rec.Host, rec.Channel); err != nil {
+ log.Printf("provision: dry-run: would converge %s channel %s -> %s", rec.Host, existing.Channel, channel)
+ } else if err := reg.SetChannel(rec.Host, channel); err != nil {
log.Printf("provision: converge channel on %s: %v", rec.Host, err)
} else {
- log.Printf("provision: converged %s channel %s -> %s", rec.Host, existing.Channel, rec.Channel)
+ log.Printf("provision: converged %s channel %s -> %s", rec.Host, existing.Channel, channel)
}
}
if existing.Environment != rec.Environment {
@@ -626,6 +653,19 @@ func convergeOwned(reg *registry.Registry, existing registry.Route, rec Record,
log.Printf("provision: converged %s steering %q -> %q", rec.Host, existing.Steering, rec.Steering)
}
}
+ // The record's Home is authoritative over the row's link (#162), in both
+ // directions: a row whose link drifted from the record would have the
+ // domains pass derive an alias serving an account from a box its record
+ // no longer places it in.
+ if rec.Role == "replica" && existing.ReplicaOf != rec.Home {
+ if dryRun {
+ log.Printf("provision: dry-run: would converge %s replica_of %q -> %q", rec.Host, existing.ReplicaOf, rec.Home)
+ } else if err := reg.SetReplicaOf(rec.Host, rec.Home); err != nil {
+ log.Printf("provision: converge replica_of on %s: %v", rec.Host, err)
+ } else {
+ log.Printf("provision: converged %s replica_of %q -> %q", rec.Host, existing.ReplicaOf, rec.Home)
+ }
+ }
}
// Reconcile materializes bucket instance records into registry routes for
@@ -653,6 +693,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
return fmt.Errorf("provision: list apps: %w", err)
}
recs := map[string]Record{} // host -> record, in-scope only
+ all := map[string]Record{} // host -> record, EVERY decoded record
// protect holds hosts whose record KEY parsed (ParseRecordKey succeeded)
// but whose body could not be trusted — unparseable JSON, or a body that
// disagrees with its own path. Such a host must NEVER be treated as "the
@@ -728,6 +769,16 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
protect[host] = true
continue
}
+ // Every decoded record, in scope or not: a replica's box needs its
+ // HOME record (whose Region is another box) for the route's
+ // channel, and the siblings pass needs both sides. Scope is about
+ // what this box MINTS, never about what it may read.
+ //
+ // Collected HERE, after the shape and channel rails and before the
+ // scope check: a record ValidateShape rejected, or one carrying no
+ // channel, is absent from this map too, so a replica pointed at it
+ // is protected rather than minted off a record no box will serve.
+ all[rec.Host] = rec
// Scope. For a home-boxed record this is the region/pool test it
// always was; for a static or public-addr one the region half widens
// to every platform box in the deployment (everywhereClass, above).
@@ -745,6 +796,9 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
}
recs[rec.Host] = rec
}
+ if cfg.OnRecords != nil {
+ cfg.OnRecords(all)
+ }
// --- manifest seed cache, per (acct, app), lazily -------------------------
// known=false (a genuine read failure, not "no manifest") means the seed
@@ -836,6 +890,20 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
}
for _, rec := range recs {
+ // A replica's route takes its channel from its HOME record (#162),
+ // so a pass that cannot see that record can neither mint the row nor
+ // converge it: both would write an empty channel off a zero Record.
+ // Protect rather than guess — the row keeps serving what it serves,
+ // and the removal phase leaves it alone (spec §2). Asked before
+ // trashedFor and seedFor, so a mirror of nothing costs no reads.
+ if rec.Role == "replica" {
+ if _, ok := all[rec.Home]; !ok {
+ log.Printf("provision: not minting replica %s: its home record %s is absent or unreadable this pass",
+ rec.Host, rec.Home)
+ protect[rec.Host] = true
+ continue
+ }
+ }
if trashedFor(rec.Account, rec.App) {
continue
}
@@ -868,7 +936,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// exact route this branch hardcoded before they existed — that is
// the additivity rule, pinned by TestRecordWireShapeIsAdditive and
// TestReconcileMints.
- nr := routeFor(rec, cfg.DataDir, seed)
+ nr := routeFor(rec, all[rec.Home], cfg.DataDir, seed)
if cfg.DryRun {
log.Printf("provision: dry-run: would mint %s route %s (app %q acct %q channel %q)",
nr.Kind, rec.Host, rec.App, rec.Account, rec.Channel)
@@ -885,7 +953,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// deploy's SetVersions or the activator's SetNextDue would
// clobber those, and Upsert cannot express "clear Environment"
// at all (it preserves a non-empty value by design).
- convergeOwned(reg, existing, rec, cfg.DryRun)
+ convergeOwned(reg, existing, rec, all[rec.Home], cfg.DryRun)
// Absent means "this record does not speak to hibernation" —
// never a derivation stamped over an operator's own setting.
//
@@ -944,7 +1012,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// widened to prevent. What remains are the columns nothing
// converges in place: the route CLASS and the two filesystem paths
// derived from it.
- if want := routeFor(rec, cfg.DataDir, seed); want.Kind != existing.Kind ||
+ if want := routeFor(rec, all[rec.Home], cfg.DataDir, seed); want.Kind != existing.Kind ||
(rec.Unit != "" && want.Unit != existing.Unit) ||
want.Socket != existing.Socket || want.DBPath != existing.DBPath {
log.Printf("provision: %s: record describes a different route SHAPE than the row "+
@@ -962,7 +1030,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// as before: adoption grants this pass's removal phase authority
// over the row, so the match test is the only thing standing
// between a backfill and a retired production route.
- want := routeFor(rec, cfg.DataDir, seed)
+ want := routeFor(rec, all[rec.Home], cfg.DataDir, seed)
if reason := shapeMismatch(existing, want); reason != "" {
log.Printf("provision: %s exists but is not the route this record describes (%s) — leaving it alone",
rec.Host, reason)
@@ -979,7 +1047,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// already on the record's channel. Every write inside
// convergeOwned is dry-run guarded there, so this call
// writes nothing.
- convergeOwned(reg, existing, rec, true)
+ convergeOwned(reg, existing, rec, all[rec.Home], true)
continue
}
if err := reg.SetProvisioned(rec.Host, true); err != nil {
@@ -994,7 +1062,7 @@ func Reconcile(ctx context.Context, s release.Store, reg *registry.Registry, cfg
// live process is allowed to sleep. It converges on the next
// pass through the owned branch above, by which point the
// adoption is already visible in the log.
- convergeOwned(reg, existing, rec, cfg.DryRun)
+ convergeOwned(reg, existing, rec, all[rec.Home], cfg.DryRun)
}
}
diff --git a/internal/provision/reconcile_test.go b/internal/provision/reconcile_test.go
index e524ad2fb..edf6991fb 100644
--- a/internal/provision/reconcile_test.go
+++ b/internal/provision/reconcile_test.go
@@ -922,7 +922,7 @@ func TestReconcileConvergesLegacyAddrToAddrs(t *testing.T) {
// (shapeMismatch is the adoption gate AND the drift test — the two failures
// the un-cleared addr caused, asserted at their source).
seed := seedJSON(&Manifest{Seed: []string{"seed"}}) // setup()'s manifest
- if reason := shapeMismatch(after, routeFor(rec, cfg.DataDir, seed)); reason != "" {
+ if reason := shapeMismatch(after, routeFor(rec, Record{}, cfg.DataDir, seed)); reason != "" {
t.Errorf("converged row still diverges from its record (%s) — the pass would log this every tick and refuse re-adoption", reason)
}
@@ -2505,7 +2505,7 @@ func TestEverywhereClassAgreesAcrossRecordAndRow(t *testing.T) {
t.Fatalf("fixture is not a shape this pass would ever see: %v", err)
}
fromRecord := everywhereClass(f.rec)
- fromRow := everywhereRoute(routeFor(f.rec, "/data", `["seed"]`))
+ fromRow := everywhereRoute(routeFor(f.rec, Record{}, "/data", `["seed"]`))
if fromRecord != fromRow {
t.Fatalf("class disagrees across the mint and removal gates: everywhereClass = %v, "+
"everywhereRoute(routeFor(...)) = %v — a record this box mints must be a row it can retire",
@@ -2524,7 +2524,7 @@ func TestEverywhereClassAgreesAcrossRecordAndRow(t *testing.T) {
func TestRouteForCarriesAddrsAndHealth(t *testing.T) {
rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable",
Kind: "service", Addrs: []string{"10.0.0.1:3000", "10.0.0.2:3000"}}
- rt := routeFor(rec, "/data", "seed")
+ rt := routeFor(rec, Record{}, "/data", "seed")
if rt.Addrs != "10.0.0.1:3000,10.0.0.2:3000" {
t.Fatalf("Addrs = %q", rt.Addrs)
}
@@ -2537,7 +2537,7 @@ func TestRouteForCarriesAddrsAndHealth(t *testing.T) {
}
// Legacy single-addr record: byte-identical route to before this change.
old := Record{Host: "s.example.com", App: "web", Account: "bab", Channel: "stable", Kind: "service", Addr: "10.0.0.9:3000"}
- ort := routeFor(old, "/data", "seed")
+ ort := routeFor(old, Record{}, "/data", "seed")
if ort.Addrs != "" || ort.Health != "" || ort.Addr != "10.0.0.9:3000" {
t.Fatalf("legacy record's route changed shape: %+v", ort)
}
@@ -2553,7 +2553,7 @@ func TestRouteForMaterializesPartialHealthDefaults(t *testing.T) {
rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable",
Kind: "service", Addrs: []string{"10.0.0.1:3000", "10.0.0.2:3000"},
Health: &Health{Path: "/healthz"}}
- rt := routeFor(rec, "/data", "seed")
+ rt := routeFor(rec, Record{}, "/data", "seed")
var h Health
if err := json.Unmarshal([]byte(rt.Health), &h); err != nil {
t.Fatalf("health JSON = %q (%v)", rt.Health, err)
@@ -2701,11 +2701,11 @@ func TestReconcileUnsetInboundMintsAsBefore(t *testing.T) {
// onto the route, byte for byte — no derivation, unlike Hibernate/StatusNamed.
func TestRouteForCarriesSteering(t *testing.T) {
rec := Record{Host: "m.example.com", App: "web", Account: "bab", Channel: "stable", Steering: "latency"}
- if got := routeFor(rec, "/data", "seed").Steering; got != "latency" {
+ if got := routeFor(rec, Record{}, "/data", "seed").Steering; got != "latency" {
t.Fatalf("Steering = %q, want latency", got)
}
unsteered := Record{Host: "s.example.com", App: "web", Account: "bab", Channel: "stable"}
- if got := routeFor(unsteered, "/data", "seed").Steering; got != "" {
+ if got := routeFor(unsteered, Record{}, "/data", "seed").Steering; got != "" {
t.Fatalf("Steering = %q, want empty for a record that never mentioned it", got)
}
}
@@ -2735,3 +2735,120 @@ func TestShapeMismatchIgnoresSteering(t *testing.T) {
t.Fatal("a genuine kind mismatch was hidden behind the ignored steering column")
}
}
+
+// --- replica placements (#162) -------------------------------------------
+
+// replicaSetup writes an account's HOME record (region ie, steering latency,
+// one replica placement in uk) and its REPLICA record (region uk, role
+// replica, home = the account's public host) into one store. Every test
+// below runs a pass as the uk box, the ie box, or both, over this pair.
+func replicaSetup(t *testing.T) (*release.DirStore, string, Record, Record) {
+ t.Helper()
+ s, sqid := setup(t)
+ home := acmeRecord(sqid)
+ home.Region, home.Steering, home.Replicas = "ie", "latency", []string{"uk"}
+ if err := CreateRecord(ctx, s, home); err != nil {
+ t.Fatal(err)
+ }
+ replica := acmeRecord(sqid)
+ replica.Host, replica.Slug = "r7x2k9m4qa.go.tito.io", "r7x2k9m4qa"
+ replica.Role, replica.Home, replica.Region = "replica", home.Host, "uk"
+ replica.Replicas, replica.Steering = nil, ""
+ if err := CreateRecord(ctx, s, replica); err != nil {
+ t.Fatal(err)
+ }
+ return s, sqid, home, replica
+}
+
+func TestReconcileMintsReplicaRouteOnItsOwnBoxOnly(t *testing.T) {
+ s, _, _, replica := replicaSetup(t)
+ uk := regWith(t)
+ if err := Reconcile(ctx, s, uk, ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
+ t.Fatal(err)
+ }
+ rt := mustGet(t, uk, replica.Host)
+ if rt.ReplicaOf != "acme.go.tito.io" {
+ t.Errorf("ReplicaOf = %q, want acme.go.tito.io", rt.ReplicaOf)
+ }
+ if rt.Channel != "stable" {
+ t.Errorf("channel = %q, want the home's stable", rt.Channel)
+ }
+ ie := regWith(t)
+ if err := Reconcile(ctx, s, ie, ReconcileConfig{Region: "ie", Pools: []string{"tito"}, DataDir: "/data"}); err != nil {
+ t.Fatal(err)
+ }
+ // Get reports a miss as (nil, nil), never an error — the absence has to
+ // be read off the route, not off err.
+ if rt, err := ie.Get(replica.Host); err != nil {
+ t.Fatal(err)
+ } else if rt != nil {
+ t.Errorf("the home's box minted a route for the replica: %+v", rt)
+ }
+ if got := mustGet(t, ie, "acme.go.tito.io"); got.ReplicaOf != "" {
+ t.Errorf("the home's own row carries ReplicaOf %q, want empty", got.ReplicaOf)
+ }
+}
+
+func TestReconcileReplicaFollowsHomeChannel(t *testing.T) {
+ s, sqid, home, replica := replicaSetup(t)
+ uk := regWith(t)
+ cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}
+ if err := Reconcile(ctx, s, uk, cfg); err != nil {
+ t.Fatal(err)
+ }
+ home.Channel = "canary"
+ putReplace(t, s, RecordKey(sqid, "tito", home.Host), mustJSON(t, home))
+ if err := Reconcile(ctx, s, uk, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if got := mustGet(t, uk, replica.Host).Channel; got != "canary" {
+ t.Errorf("channel = %q, want canary: the replica must follow its home", got)
+ }
+}
+
+// A replica whose home record has gone is a mirror of nothing. The route it
+// already has keeps serving exactly what it was serving — it must not be
+// converged onto the zero record's empty channel — and on a box that has
+// not minted it yet, nothing is minted at all rather than minted on a guess.
+func TestReconcileProtectsReplicaWhoseHomeIsGone(t *testing.T) {
+ s, sqid, home, replica := replicaSetup(t)
+ uk := regWith(t)
+ cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data"}
+ if err := Reconcile(ctx, s, uk, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if err := DeleteRecord(ctx, s, sqid, "tito", home.Host); err != nil {
+ t.Fatal(err)
+ }
+ if err := Reconcile(ctx, s, uk, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if got := mustGet(t, uk, replica.Host).Channel; got != "stable" {
+ t.Errorf("channel = %q, want stable: the replica's row was converged off a home record that is gone", got)
+ }
+ fresh := regWith(t)
+ if err := Reconcile(ctx, s, fresh, cfg); err != nil {
+ t.Fatal(err)
+ }
+ if rt, err := fresh.Get(replica.Host); err != nil {
+ t.Fatal(err)
+ } else if rt != nil {
+ t.Errorf("a replica route was minted with no home record to take a channel from: %+v", rt)
+ }
+}
+
+func TestReconcileReportsEveryRecord(t *testing.T) {
+ s, _, home, replica := replicaSetup(t)
+ var seen map[string]Record
+ cfg := ReconcileConfig{Region: "uk", Pools: []string{"tito"}, DataDir: "/data",
+ OnRecords: func(all map[string]Record) { seen = all }}
+ if err := Reconcile(ctx, s, regWith(t), cfg); err != nil {
+ t.Fatal(err)
+ }
+ if _, ok := seen[home.Host]; !ok {
+ t.Error("OnRecords did not carry the out-of-scope home record")
+ }
+ if _, ok := seen[replica.Host]; !ok {
+ t.Error("OnRecords did not carry the in-scope replica record")
+ }
+}