amadan

carlos / platform Public

edges: with region HA on, a relocatable host is home only on the box its claim names, stamped

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 089decb99c996bfc43fc5655fa3e6b0bdee82f26 parent 5698cbc
4 files changed, +603 −6
  • internal/edge/homemap_alias_stamp_test.go +18 −0
  • internal/host/edgesclaims.go +225 −0
  • internal/host/edgesclaims_test.go +327 −0
  • internal/host/edgesfile.go +33 −6
diff --git a/internal/edge/homemap_alias_stamp_test.go b/internal/edge/homemap_alias_stamp_test.go
new file mode 100644
index 000000000..d5cab3d98
--- /dev/null
+++ b/internal/edge/homemap_alias_stamp_test.go
@@ -0,0 +1,18 @@
+package edge
+
+import "testing"
+
+// Region HA plan 1b (Astra plan R1-2): after a failover the dead box's file
+// is stale but still listed. Its alias carries the old installation's
+// stamp and the successor's the new one, so the successor wins however the
+// boxes sort.
+func TestHomeMapAnAliasFollowsTheNewerInstallation(t *testing.T) {
+ s := meshTestStore(t)
+ putJSON(t, s, "control/edges/annie.json", `{"v":1,"box":"annie","eip":"1.1.1.1","hosts":{"shop.test":{"role":"home","stamp":41},"www.customer.com":{"role":"home","alias":true,"stamp":41}}}`)
+ putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"shop.test":{"role":"home","stamp":57},"www.customer.com":{"role":"home","alias":true,"stamp":57}}}`)
+ m := newHomeMap(s, "carol")
+ refreshHomeMap(t, m)
+ if eip, box, ok := m.Home("www.customer.com"); !ok || box != "betty" || eip != "2.2.2.2" {
+ t.Fatalf("Home(www.customer.com) = %q %q %v, want betty", eip, box, ok)
+ }
+}
diff --git a/internal/host/edgesclaims.go b/internal/host/edgesclaims.go
new file mode 100644
index 000000000..fda76f13f
--- /dev/null
+++ b/internal/host/edgesclaims.go
@@ -0,0 +1,225 @@
+package host
+
+import (
+ "context"
+ "math"
+ "sort"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// Region HA spec 1 §5.4: with CARLOS_REGION_HA on, a relocatable host is
+// published as home only by the box its claim's op.exec names, carrying
+// the claim's holder.stamp so the edge's home map believes the newest
+// installation (HA4). The edge never reads claims; this publisher does, on
+// a slow cache, because the safety gate is the activator's fresh read at
+// every spawn (claim C4), not this file.
+
+const (
+ // edgesClaimTTL is how long one claim read stands: reconcile's sweep
+ // (H9). A held row's entry can be stale this long, which can only send
+ // a request to a box that then refuses to spawn.
+ edgesClaimTTL = 10 * time.Minute
+ // edgesClaimReadsPerPass bounds the GETs one pass makes, so a box with
+ // thousands of hosts spreads its reads over several passes instead of
+ // stalling one. A host not read yet publishes as it did before this
+ // file existed.
+ edgesClaimReadsPerPass = 256
+ // edgesClaimPassEvery is how often the loop runs an extra pass while
+ // claim reads are due. The registry timer alone (5 to 6 minutes) would
+ // take 40 minutes to refresh 2,000 hosts at 256 a pass; with these
+ // passes every host is re-read within about edgesClaimTTL plus four
+ // minutes.
+ edgesClaimPassEvery = 30 * time.Second
+)
+
+// edgesDecision is what one claim says about this box and one host.
+type edgesDecision struct {
+ withhold bool // the claim names another executor, or none
+ stamp int64 // holder.stamp when this box is the executor
+}
+
+// edgesClaimEntry is one host's last decision and last read attempt. A
+// failed read moves attempted and keeps the decision, so a claim that
+// cannot be read goes to the back of the queue instead of holding its
+// place at the front (claim C6: an error is never absence).
+type edgesClaimEntry struct {
+ decision edgesDecision
+ decided bool // some read of this claim has succeeded
+ attempted time.Time
+}
+
+// edgesClaims caches one decision per host. It is used only by the edges
+// file loop's goroutine.
+type edgesClaims struct {
+ self string // this box's reference: the edges file's box_id
+ read func(ctx context.Context, host string) (hostclaim.Observed, error)
+ now func() time.Time
+ entries map[string]edgesClaimEntry
+ hosts []string // the relocatable hosts of the last pass
+}
+
+func newEdgesClaims(self string, read func(context.Context, string) (hostclaim.Observed, error)) *edgesClaims {
+ return &edgesClaims{self: self, read: read, now: time.Now, entries: map[string]edgesClaimEntry{}}
+}
+
+// relocatableRow is spec 1 §2's relocatable host as a registry row shows
+// it: a provisioned instance whose process is the activator's own exec
+// child. Unit-backed, addr-backed, static and service routes keep their
+// box and are published as before; sidecars are never published at all.
+func relocatableRow(r registry.Route) bool {
+ return r.Provisioned && r.Kind == "instance" && r.Backing == "" && r.Unit == "" &&
+ r.Addr == "" && r.Addrs == "" && r.Socket != ""
+}
+
+// decideClaim turns one observation into a decision. An absent claim
+// leaves the host as it was before claims existed: a region can only be
+// given a second box once every host has one (spec §8).
+func (c *edgesClaims) decideClaim(obs hostclaim.Observed) edgesDecision {
+ if obs.Absent || obs.Claim == nil {
+ return edgesDecision{}
+ }
+ cl := obs.Claim
+ if cl.Holder == nil || cl.Op.Exec != c.self {
+ return edgesDecision{withhold: true}
+ }
+ stamp := cl.Holder.Stamp
+ if stamp > math.MaxInt64 {
+ stamp = 0
+ }
+ return edgesDecision{stamp: int64(stamp)}
+}
+
+func (c *edgesClaims) dueAt(h string, now time.Time) bool {
+ e, ok := c.entries[h]
+ return !ok || now.Sub(e.attempted) >= edgesClaimTTL
+}
+
+// backlog reports whether any host of the last pass has a read due.
+func (c *edgesClaims) backlog(now time.Time) bool {
+ for _, h := range c.hosts {
+ if c.dueAt(h, now) {
+ return true
+ }
+ }
+ return false
+}
+
+// decide returns a decision for each host that has one. Hosts whose last
+// attempt is older than edgesClaimTTL are read, never-attempted first and
+// then oldest attempt first, up to edgesClaimReadsPerPass. A read that
+// fails keeps the previous decision; a host with no successful read has
+// none, and publishes as before.
+func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edgesDecision {
+ now := c.now()
+ c.hosts = append(c.hosts[:0], hosts...)
+ want := make(map[string]bool, len(hosts))
+ var due []string
+ for _, h := range hosts {
+ want[h] = true
+ if c.dueAt(h, now) {
+ due = append(due, h)
+ }
+ }
+ for h := range c.entries {
+ if !want[h] {
+ delete(c.entries, h)
+ }
+ }
+ sort.Slice(due, func(i, j int) bool {
+ ei, iok := c.entries[due[i]]
+ ej, jok := c.entries[due[j]]
+ if iok != jok {
+ return !iok // never attempted first
+ }
+ if !ei.attempted.Equal(ej.attempted) {
+ return ei.attempted.Before(ej.attempted)
+ }
+ return due[i] < due[j]
+ })
+ if len(due) > edgesClaimReadsPerPass {
+ due = due[:edgesClaimReadsPerPass]
+ }
+ for _, h := range due {
+ e := c.entries[h]
+ e.attempted = now
+ if obs, err := c.read(ctx, h); err == nil {
+ e.decision, e.decided = c.decideClaim(obs), true
+ }
+ c.entries[h] = e
+ }
+ out := make(map[string]edgesDecision, len(c.entries))
+ for h, e := range c.entries {
+ if e.decided {
+ out[h] = e.decision
+ }
+ }
+ return out
+}
+
+// applyEdgesClaims withholds or stamps every relocatable row in hosts
+// (edgesFileWrite's served map, before aliases join it, so an alias of a
+// withheld host is withheld with it and an alias of a stamped host takes
+// its stamp). claims nil leaves hosts untouched.
+func applyEdgesClaims(ctx context.Context, claims *edgesClaims, routes []registry.Route, hosts map[string]edgeHost) {
+ if claims == nil {
+ return
+ }
+ var reloc []string
+ for _, r := range routes {
+ if _, served := hosts[r.Host]; served && relocatableRow(r) {
+ reloc = append(reloc, r.Host)
+ }
+ }
+ for host, d := range claims.decide(ctx, reloc) {
+ h, ok := hosts[host]
+ if !ok {
+ continue
+ }
+ if d.withhold {
+ delete(hosts, host)
+ continue
+ }
+ h.Stamp = d.stamp
+ hosts[host] = h
+ }
+}
+
+// edgesLoopState is edgesFileLoop's per-tick decision: the registry and
+// timer schedule, plus, with region HA on, an extra pass every
+// edgesClaimPassEvery while claim reads are due. A claim pass is a
+// publication like any other: it waits out the schedule's retry backoff
+// after a failed write, and its own outcome moves that backoff, so a bucket
+// that refuses every write is not asked every 30 seconds (Astra plan R2-1).
+type edgesLoopState struct {
+ sched *edgesFileSchedule
+ claims *edgesClaims
+ claimsNext time.Time
+ schedDue bool
+ claimDue bool
+}
+
+// due reports whether to run a pass on this tick.
+func (s *edgesLoopState) due(now time.Time, fp string, fpOK bool) bool {
+ s.schedDue = s.sched.due(now, fp, fpOK)
+ s.claimDue = s.claims != nil && !now.Before(s.claimsNext) && !now.Before(s.sched.retryAt) && s.claims.backlog(now)
+ return s.schedDue || s.claimDue
+}
+
+// timed reports whether the pass due just allowed is the timer's.
+func (s *edgesLoopState) timed() bool { return s.schedDue && s.sched.timed }
+
+// done records the pass's outcome. Every pass, claim passes included, goes
+// through the schedule: a failure advances its backoff, and a success
+// clears it and records the fingerprint the pass published. A claim pass
+// is never the timer's (due cleared timed), so the timer does not move.
+func (s *edgesLoopState) done(now time.Time, ok bool) {
+ if s.schedDue || s.claimDue {
+ s.sched.done(now, ok)
+ }
+ if s.claims != nil {
+ s.claimsNext = now.Add(edgesClaimPassEvery)
+ }
+}
diff --git a/internal/host/edgesclaims_test.go b/internal/host/edgesclaims_test.go
new file mode 100644
index 000000000..8b36e9aee
--- /dev/null
+++ b/internal/host/edgesclaims_test.go
@@ -0,0 +1,327 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// seedInstance upserts a ready, provisioned exec instance: a relocatable
+// host (spec 1 §2).
+func seedInstance(t *testing.T, reg *registry.Registry, host string) {
+ t.Helper()
+ if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", Socket: "/run/carlos/" + host + ".sock",
+ Provisioned: true, App: "jam", Account: "bab", Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// claimsBook answers claim reads from a map; a missing host is an absent
+// claim, a host in errs fails.
+type claimsBook struct {
+ claims map[string]*hostclaim.Claim
+ errs map[string]bool
+ reads []string
+}
+
+func (b *claimsBook) read(_ context.Context, host string) (hostclaim.Observed, error) {
+ b.reads = append(b.reads, host)
+ if b.errs[host] {
+ return hostclaim.Observed{}, errors.New("SlowDown")
+ }
+ c, ok := b.claims[host]
+ if !ok {
+ return hostclaim.Observed{Host: host, Absent: true}, nil
+ }
+ return hostclaim.Observed{Host: host, Claim: c, ETag: "e"}, nil
+}
+
+func serving(exec string, stamp uint64) *hostclaim.Claim {
+ return &hostclaim.Claim{V: 2, Gen: 7, Holder: &hostclaim.Holder{Stamp: stamp},
+ Op: hostclaim.Op{Kind: hostclaim.KindServing, ID: "op-1", Exec: exec}}
+}
+
+func publishWithClaims(t *testing.T, reg *registry.Registry, claims *edgesClaims) map[string]edgeHost {
+ t.Helper()
+ st := release.NewDirStore(t.TempDir())
+ var last []byte
+ if err := edgesFileWriteClaims(context.Background(), st, reg, edgesFile{Box: "annie", BoxID: "annie"}, "1.2.3.4", &last, false, claims); err != nil {
+ t.Fatal(err)
+ }
+ body, _, err := st.Get(context.Background(), edgesFileKey("annie"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var f edgesFile
+ if err := json.Unmarshal(body, &f); err != nil {
+ t.Fatal(err)
+ }
+ return f.Hosts
+}
+
+func TestEdgesFileFollowsTheClaimsExecutor(t *testing.T) {
+ reg := edgesTestRegistry(t)
+ for _, h := range []string{"mine.bab.test", "theirs.bab.test", "between.bab.test", "unclaimed.bab.test"} {
+ seedInstance(t, reg, h)
+ }
+ seedRoute(t, reg, "static.example.com") // not relocatable: never read, always home
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{
+ "mine.bab.test": serving("annie", 41),
+ "theirs.bab.test": serving("bert", 42),
+ "between.bab.test": serving("", 43), // op.exec none: nobody serves it
+ }}
+ hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
+ if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 || h.Role != "home" {
+ t.Fatalf("executor's host = %+v, %v; want home stamped 41", h, ok)
+ }
+ for _, h := range []string{"theirs.bab.test", "between.bab.test"} {
+ if _, ok := hosts[h]; ok {
+ t.Errorf("%s published by a box its claim does not name", h)
+ }
+ }
+ if h, ok := hosts["unclaimed.bab.test"]; !ok || h.Stamp != 0 {
+ t.Errorf("a host with no claim = %+v, %v; want home, unstamped, as before", h, ok)
+ }
+ if h, ok := hosts["static.example.com"]; !ok || h.Stamp != 0 {
+ t.Errorf("a service route = %+v, %v; want home, unstamped", h, ok)
+ }
+ for _, r := range book.reads {
+ if r == "static.example.com" {
+ t.Error("read a claim for a route that is not relocatable")
+ }
+ }
+}
+
+func TestEdgesFileWithholdsTheAliasOfAWithheldHost(t *testing.T) {
+ reg := edgesTestRegistry(t)
+ seedInstance(t, reg, "theirs.bab.test")
+ if err := reg.UpsertAlias(registry.RouteAlias{Alias: "www.customer.com", Host: "theirs.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"}); err != nil {
+ t.Fatal(err)
+ }
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"theirs.bab.test": serving("bert", 42)}}
+ hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
+ if len(hosts) != 0 {
+ t.Fatalf("hosts = %v, want none", hosts)
+ }
+}
+
+// Review Focus 2: a claim that cannot be read never moves a host. It keeps
+// the last decision, or, never read, publishes as it did before claims.
+func TestEdgesFileKeepsTheLastDecisionWhenAClaimCannotBeRead(t *testing.T) {
+ reg := edgesTestRegistry(t)
+ seedInstance(t, reg, "mine.bab.test")
+ seedInstance(t, reg, "theirs.bab.test")
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41), "theirs.bab.test": serving("bert", 42)}}
+ clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ claims := newEdgesClaims("annie", book.read)
+ claims.now = func() time.Time { return clock }
+ publishWithClaims(t, reg, claims)
+ clock = clock.Add(edgesClaimTTL)
+ book.errs = map[string]bool{"mine.bab.test": true, "theirs.bab.test": true}
+ hosts := publishWithClaims(t, reg, claims)
+ if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 {
+ t.Fatalf("mine = %+v, %v: an unreadable claim dropped or unstamped a served host", h, ok)
+ }
+ if _, ok := hosts["theirs.bab.test"]; ok {
+ t.Fatal("an unreadable claim published a host this box was told not to")
+ }
+
+ fresh := newEdgesClaims("annie", book.read)
+ if h, ok := publishWithClaims(t, reg, fresh)["mine.bab.test"]; !ok || h.Stamp != 0 {
+ t.Fatalf("never read = %+v, %v; want home, unstamped", h, ok)
+ }
+}
+
+func TestEdgesClaimsRereadOnlyAfterTheTTL(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41)}}
+ clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ c := newEdgesClaims("annie", book.read)
+ c.now = func() time.Time { return clock }
+ ctx := context.Background()
+ c.decide(ctx, []string{"mine.bab.test"})
+ clock = clock.Add(edgesClaimTTL - time.Second)
+ c.decide(ctx, []string{"mine.bab.test"})
+ if len(book.reads) != 1 {
+ t.Fatalf("%d reads inside the TTL, want 1", len(book.reads))
+ }
+ book.claims["mine.bab.test"] = serving("bert", 50)
+ clock = clock.Add(time.Second)
+ if d := c.decide(ctx, []string{"mine.bab.test"})["mine.bab.test"]; !d.withhold {
+ t.Fatalf("decision after the TTL = %+v; the move to another box was not seen", d)
+ }
+ c.decide(ctx, nil)
+ if len(c.entries) != 0 {
+ t.Fatalf("a host no longer served kept its entry: %v", c.entries)
+ }
+}
+
+func TestEdgesClaimsBoundReadsPerPassNeverReadFirst(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
+ clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ c := newEdgesClaims("annie", book.read)
+ c.now = func() time.Time { return clock }
+ var hosts []string
+ for i := 0; i < edgesClaimReadsPerPass+10; i++ {
+ hosts = append(hosts, fmt.Sprintf("h%04d.bab.test", i))
+ }
+ ctx := context.Background()
+ c.decide(ctx, hosts)
+ if len(book.reads) != edgesClaimReadsPerPass {
+ t.Fatalf("first pass read %d claims, want %d", len(book.reads), edgesClaimReadsPerPass)
+ }
+ book.reads = nil
+ c.decide(ctx, hosts)
+ if len(book.reads) != 10 {
+ t.Fatalf("second pass read %d claims, want the 10 never read", len(book.reads))
+ }
+}
+
+// Astra plan R1-2: an alias and a served catch-all are dialled through
+// their primary's installation, so they carry its stamp. Unstamped, a dead
+// predecessor's alias would tie with the successor's and the edge would
+// keep the first box by name.
+func TestEdgesFileStampsAliasesAndCatchAllsWithTheirPrimary(t *testing.T) {
+ reg := edgesTestRegistry(t)
+ seedInstance(t, reg, "mine.bab.test")
+ for _, a := range []registry.RouteAlias{
+ {Alias: "www.customer.com", Host: "mine.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"},
+ {Alias: "*.customer.com", Host: "mine.bab.test", Account: "bab", App: "jam", DNSStatus: "pointing"},
+ } {
+ if err := reg.UpsertAlias(a); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if err := reg.UpsertWildcardDomain(registry.WildcardDomain{Host: "customer.com", Account: "bab", App: "jam"}); err != nil {
+ t.Fatal(err)
+ }
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 57)}}
+ hosts := publishWithClaims(t, reg, newEdgesClaims("annie", book.read))
+ for _, key := range []string{"mine.bab.test", "www.customer.com", "*.customer.com"} {
+ if h, ok := hosts[key]; !ok || h.Stamp != 57 || h.Role != "home" {
+ t.Errorf("%s = %+v, %v; want home at stamp 57", key, h, ok)
+ }
+ }
+ // With no claim reader the file is what it always was: no stamps.
+ for key, h := range publishWithClaims(t, reg, nil) {
+ if h.Stamp != 0 {
+ t.Errorf("%s stamped %d with region HA off", key, h.Stamp)
+ }
+ }
+}
+
+// Astra plan R1-4: claims that keep failing go to the back of the queue,
+// so they cannot hold every read of every pass.
+func TestEdgesClaimsAFailingClaimDoesNotStarveTheRest(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"zz-healthy.bab.test": serving("annie", 7)}, errs: map[string]bool{}}
+ var hosts []string
+ for i := 0; i < edgesClaimReadsPerPass; i++ {
+ h := fmt.Sprintf("a%04d.bab.test", i)
+ book.errs[h] = true
+ hosts = append(hosts, h)
+ }
+ hosts = append(hosts, "zz-healthy.bab.test")
+ clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ c := newEdgesClaims("annie", book.read)
+ c.now = func() time.Time { return clock }
+ ctx := context.Background()
+ c.decide(ctx, hosts) // the 256 failing claims sort first and all fail
+ clock = clock.Add(edgesClaimPassEvery)
+ if d, ok := c.decide(ctx, hosts)["zz-healthy.bab.test"]; !ok || d.stamp != 7 {
+ t.Fatalf("healthy claim = %+v, %v: starved by claims that cannot be read", d, ok)
+ }
+}
+
+// Astra plan R1-5: the loop's own schedule, with a stable registry, keeps
+// every claim of a 2,000-host box fresh. Timer passes alone come every
+// five to six minutes and read 256 claims each.
+func TestEdgesLoopRefreshesEveryClaimWithinTheTTL(t *testing.T) {
+ const n = 2000
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
+ var hosts []string
+ for i := 0; i < n; i++ {
+ h := fmt.Sprintf("h%04d.bab.test", i)
+ book.claims[h] = serving("annie", uint64(i+1))
+ hosts = append(hosts, h)
+ }
+ start := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ clock := start
+ claims := newEdgesClaims("annie", book.read)
+ claims.now = func() time.Time { return clock }
+ state := &edgesLoopState{sched: newEdgesFileSchedule(start, func() time.Duration { return 5*time.Minute + 30*time.Second }), claims: claims}
+ lastRead := map[string]time.Time{}
+ worst := time.Duration(0)
+ for clock = start; clock.Before(start.Add(3 * time.Hour)); clock = clock.Add(edgesFileWatchInterval) {
+ if !state.due(clock, "fp", true) {
+ continue
+ }
+ book.reads = nil
+ claims.decide(context.Background(), hosts)
+ for _, h := range book.reads {
+ lastRead[h] = clock
+ }
+ state.done(clock, true)
+ if clock.Sub(start) < time.Hour {
+ continue // warm-up: the first reads of every host
+ }
+ for _, h := range hosts {
+ if age := clock.Sub(lastRead[h]); age > worst {
+ worst = age
+ }
+ }
+ }
+ if limit := edgesClaimTTL + 5*time.Minute; worst > limit {
+ t.Fatalf("a claim went %s without a read, want at most %s", worst, limit)
+ }
+}
+
+// Astra plan R2-1: a claim pass is a publication. While the bucket refuses
+// every write, claim passes wait out the schedule's backoff and their own
+// failures move it; once writes succeed, the 30-second claim passes resume.
+func TestEdgesLoopClaimPassesObeyTheRetryBackoff(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{}}
+ var hosts []string
+ for i := 0; i < 4000; i++ { // enough that reads stay due throughout
+ h := fmt.Sprintf("h%04d.bab.test", i)
+ book.claims[h] = serving("annie", 1)
+ hosts = append(hosts, h)
+ }
+ start := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ clock := start
+ claims := newEdgesClaims("annie", book.read)
+ claims.now = func() time.Time { return clock }
+ state := &edgesLoopState{sched: newEdgesFileSchedule(start, func() time.Duration { return 5*time.Minute + 30*time.Second }), claims: claims}
+ failUntil := start.Add(20 * time.Minute)
+ var during, after []time.Time
+ for clock = start; clock.Before(failUntil.Add(8 * time.Minute)); clock = clock.Add(edgesFileWatchInterval) {
+ if !state.due(clock, "fp", true) {
+ continue
+ }
+ claims.decide(context.Background(), hosts)
+ ok := !clock.Before(failUntil)
+ if ok {
+ after = append(after, clock)
+ } else {
+ during = append(during, clock)
+ }
+ state.done(clock, ok)
+ }
+ // Unthrottled, a claim pass every 30 seconds would be 40 attempts in
+ // 20 minutes. The backoff (2 s doubling to 5 min) plus the timer allows
+ // about a dozen.
+ t.Logf("%d attempts while failing, %d after", len(during), len(after))
+ if len(during) > 16 {
+ t.Fatalf("%d publication attempts while every write failed, want at most 16: %v", len(during), during)
+ }
+ // Recovery waits out the backoff (at most five minutes), then the
+ // 30-second claim passes resume.
+ if len(after) < 3 || after[len(after)-1].Sub(after[len(after)-2]) != edgesClaimPassEvery {
+ t.Fatalf("passes after writes recovered = %v, want the 30-second claim passes back within eight minutes", after)
+ }
+}
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index be1f47275..a569d7553 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -13,6 +13,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -337,6 +338,13 @@ func edgesFilePass(ctx context.Context, s release.Store, reg *registry.Registry,
// period stale. On every other pass a headroom change alone is not a
// reason to write.
func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool) error {
+ return edgesFileWriteClaims(ctx, s, reg, ident, eip, last, refresh, nil)
+}
+
+// edgesFileWriteClaims is edgesFileWrite with the claim reader region HA
+// adds (edgesclaims.go). claims is nil unless CARLOS_REGION_HA is on, and
+// nil publishes exactly what edgesFileWrite always has.
+func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Registry, ident edgesFile, eip string, last *[]byte, refresh bool, claims *edgesClaims) error {
box := ident.Box
routes, err := reg.List()
if err != nil {
@@ -391,6 +399,10 @@ func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry
// the account apex that rides along.
hosts[r.Host] = edgeHost{Role: "home", Platform: r.Platform || r.Provisioned, Replica: r.Replica}
}
+ // Region HA (spec 1 §5.4): a relocatable host is home only on the box
+ // its claim names, stamped with the installation. Before owners and
+ // aliases, so an alias of a withheld host is withheld with it.
+ applyEdgesClaims(ctx, claims, routes, hosts)
// Owners (#175 P7): computed over the served rows alone, before any alias
// or catch-all joins the map, so an alias can only ever inherit a route's.
paths, err := reg.ListPaths()
@@ -454,7 +466,11 @@ func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry
// included, so the alias's owner is the primary's, never the alias
// row's own columns.
acct, app := owners[a.Host].owner()
- hosts[a.Alias] = edgeHost{Role: "home", Alias: true, Account: acct, App: app}
+ // The alias is served by its primary's installation, so it carries
+ // the primary's stamp (region HA HA4): an unstamped alias would tie
+ // with a dead predecessor's and the edge would keep the first box
+ // by name.
+ hosts[a.Alias] = edgeHost{Role: "home", Alias: true, Account: acct, App: app, Stamp: hosts[a.Host].Stamp}
}
// Replica placements (#162): a row minted from a role=replica record —
@@ -529,8 +545,10 @@ func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry
// mark means, so say so.
role, alias := "wildcard", false
var acct, app string
+ var stamp int64
if primary, served := catchallServed[wd.Host]; served {
role, alias = "home", true
+ stamp = hosts[primary].Stamp // the dial half is the primary's installation
// The owner is the wildcard claim's, and only when the primary
// the catch-all is served by (with its prefixes) agrees with it:
// every route behind the name has to have one owner.
@@ -538,7 +556,7 @@ func edgesFileWrite(ctx context.Context, s release.Store, reg *registry.Registry
o.add(wd.Account, wd.App)
acct, app = o.owner()
}
- hosts[key] = edgeHost{Role: role, Wildcard: true, Alias: alias, Account: acct, App: app}
+ hosts[key] = edgeHost{Role: role, Wildcard: true, Alias: alias, Account: acct, App: app, Stamp: stamp}
}
if eip == "" && *last != nil {
@@ -734,11 +752,20 @@ func edgesFileLoop(reg *registry.Registry) {
Capabilities: edgesFileCapabilities(os.Getenv),
}
dataDir := provisionTickConfig(os.Getenv).DataDir
+ // Region HA reads each relocatable host's claim (spec 1 §5.4); off, the
+ // file is exactly what it always was.
+ var claims *edgesClaims
+ if regionHAEnabled(os.Getenv) {
+ claims = newEdgesClaims(box, func(ctx context.Context, host string) (hostclaim.Observed, error) {
+ return hostclaim.Read(ctx, store, host)
+ })
+ }
log.Printf("agent: edges file loop armed (5min + jitter, and %s after a registry change, %s)", edgesFileWatchInterval, edgesFileKey(box))
last := seedEdgesFileGate(ctx, store, box)
sched := newEdgesFileSchedule(time.Now(), func() time.Duration {
return 5*time.Minute + rand.N(time.Minute)
})
+ state := &edgesLoopState{sched: sched, claims: claims}
tick := time.NewTicker(edgesFileWatchInterval)
defer tick.Stop()
fpErrs, passErrs := 0, 0
@@ -758,14 +785,14 @@ func edgesFileLoop(reg *registry.Registry) {
} else {
fpErrs = 0
}
- if !sched.due(now, fp, err == nil) {
+ if !state.due(now, fp, err == nil) {
continue
}
- if sched.timed {
+ if state.timed() {
eip = readPublicIP()
}
ident.Headroom = headroomFor(ident.Capabilities, dataDir, reg)
- err = edgesFileWrite(ctx, store, reg, ident, eip, &last, sched.timed)
+ err = edgesFileWriteClaims(ctx, store, reg, ident, eip, &last, state.timed(), claims)
if err != nil {
// Throttled: a bucket that refuses every write (an IAM Deny, an
// outage) is retried on a backoff, and each retry must not be a
@@ -777,7 +804,7 @@ func edgesFileLoop(reg *registry.Registry) {
} else {
passErrs = 0
}
- sched.done(now, err == nil)
+ state.done(now, err == nil)
}
}