amadan

carlos / platform Public

edges: claim reads have a 20s per-pass deadline, throttled failure logging, and never read a host hostclaim.Key rejects

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude b72d09ac6896c9b8a07a89465b12ac5988e07a90 parent 07c3030
2 files changed, +148 −3
  • internal/host/edgesclaims.go +54 −3
  • internal/host/edgesclaims_test.go +94 −0
diff --git a/internal/host/edgesclaims.go b/internal/host/edgesclaims.go
index fda76f13f..35a38250f 100644
--- a/internal/host/edgesclaims.go
+++ b/internal/host/edgesclaims.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "log"
"math"
"sort"
"time"
@@ -33,6 +34,13 @@ const (
// passes every host is re-read within about edgesClaimTTL plus four
// minutes.
edgesClaimPassEvery = 30 * time.Second
+ // edgesClaimPassDeadline bounds one pass's claim reads, so a store that
+ // stops answering cannot stall the edges file. A read the deadline cuts
+ // short is a failed read: the host keeps its last decision.
+ edgesClaimPassDeadline = 20 * time.Second
+ // edgesClaimLogEvery: a run of passes with failed claim reads is logged
+ // on its first pass and every this many after.
+ edgesClaimLogEvery = 60
)
// edgesDecision is what one claim says about this box and one host.
@@ -59,10 +67,15 @@ type edgesClaims struct {
now func() time.Time
entries map[string]edgesClaimEntry
hosts []string // the relocatable hosts of the last pass
+ // passDeadline bounds one pass's claim reads (edgesClaimPassDeadline).
+ passDeadline time.Duration
+ logf func(format string, a ...any)
+ failingRun int // consecutive passes with a failed claim read
}
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{}}
+ return &edgesClaims{self: self, read: read, now: time.Now, entries: map[string]edgesClaimEntry{},
+ passDeadline: edgesClaimPassDeadline, logf: log.Printf}
}
// relocatableRow is spec 1 §2's relocatable host as a registry row shows
@@ -111,7 +124,8 @@ func (c *edgesClaims) backlog(now time.Time) bool {
// 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.
+// none, and publishes as before. Once ctx ends no further read is made:
+// the hosts not reached stay due, in their place, for a later pass.
func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edgesDecision {
now := c.now()
c.hosts = append(c.hosts[:0], hosts...)
@@ -142,14 +156,27 @@ func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edg
if len(due) > edgesClaimReadsPerPass {
due = due[:edgesClaimReadsPerPass]
}
+ var failed, made int
+ var firstErr error
+ var firstHost string
for _, h := range due {
+ if ctx.Err() != nil {
+ break
+ }
e := c.entries[h]
e.attempted = now
+ made++
if obs, err := c.read(ctx, h); err == nil {
e.decision, e.decided = c.decideClaim(obs), true
+ } else {
+ if failed == 0 {
+ firstErr, firstHost = err, h
+ }
+ failed++
}
c.entries[h] = e
}
+ c.logReads(made, failed, firstHost, firstErr)
out := make(map[string]edgesDecision, len(c.entries))
for h, e := range c.entries {
if e.decided {
@@ -159,10 +186,29 @@ func (c *edgesClaims) decide(ctx context.Context, hosts []string) map[string]edg
return out
}
+// logReads logs a pass's failed claim reads, throttled: the first pass of
+// a failing run and every edgesClaimLogEvery-th after it, then once when a
+// pass's reads all succeed again.
+func (c *edgesClaims) logReads(made, failed int, host string, err error) {
+ if failed > 0 {
+ c.failingRun++
+ if c.failingRun == 1 || c.failingRun%edgesClaimLogEvery == 0 {
+ c.logf("agent: edges file: %d of %d claim reads failed (failing pass #%d), first %s: %v; those hosts keep their last decision", failed, made, c.failingRun, host, err)
+ }
+ return
+ }
+ if made > 0 && c.failingRun > 0 {
+ c.logf("agent: edges file: claim reads recovered after %d failing passes", c.failingRun)
+ c.failingRun = 0
+ }
+}
+
// 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.
+// its stamp). claims nil leaves hosts untouched. A host hostclaim.Key
+// rejects has no claim to read and publishes as before. The pass's reads
+// share one deadline, claims.passDeadline.
func applyEdgesClaims(ctx context.Context, claims *edgesClaims, routes []registry.Route, hosts map[string]edgeHost) {
if claims == nil {
return
@@ -170,9 +216,14 @@ func applyEdgesClaims(ctx context.Context, claims *edgesClaims, routes []registr
var reloc []string
for _, r := range routes {
if _, served := hosts[r.Host]; served && relocatableRow(r) {
+ if _, err := hostclaim.Key(r.Host); err != nil {
+ continue
+ }
reloc = append(reloc, r.Host)
}
}
+ ctx, cancel := context.WithTimeout(ctx, claims.passDeadline)
+ defer cancel()
for host, d := range claims.decide(ctx, reloc) {
h, ok := hosts[host]
if !ok {
diff --git a/internal/host/edgesclaims_test.go b/internal/host/edgesclaims_test.go
index 8b36e9aee..5cdbff15d 100644
--- a/internal/host/edgesclaims_test.go
+++ b/internal/host/edgesclaims_test.go
@@ -325,3 +325,97 @@ func TestEdgesLoopClaimPassesObeyTheRetryBackoff(t *testing.T) {
t.Fatalf("passes after writes recovered = %v, want the 30-second claim passes back within eight minutes", after)
}
}
+
+// Final review 4: a store that never answers cannot stall the edges file.
+// The pass has a deadline; a read it cuts short keeps the last decision as
+// any failed read does, and the reads not yet made wait for a later pass.
+func TestEdgesClaimsPassHasADeadline(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{"mine.bab.test": serving("annie", 41), "theirs.bab.test": serving("bert", 42)}}
+ hang := false
+ read := func(ctx context.Context, host string) (hostclaim.Observed, error) {
+ if hang {
+ select { // a Get that blocks until the context ends
+ case <-ctx.Done():
+ return hostclaim.Observed{}, ctx.Err()
+ case <-time.After(2 * time.Second): // so a missing deadline fails rather than hangs
+ return hostclaim.Observed{}, errors.New("no deadline")
+ }
+ }
+ return book.read(ctx, host)
+ }
+ clock := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC)
+ c := newEdgesClaims("annie", read)
+ c.now = func() time.Time { return clock }
+ c.passDeadline = 100 * time.Millisecond
+ c.logf = func(string, ...any) {}
+ routes := []registry.Route{
+ {Host: "mine.bab.test", Kind: "instance", Socket: "/run/carlos/mine.sock", Provisioned: true},
+ {Host: "theirs.bab.test", Kind: "instance", Socket: "/run/carlos/theirs.sock", Provisioned: true},
+ }
+ served := func() map[string]edgeHost {
+ return map[string]edgeHost{"mine.bab.test": {Role: "home"}, "theirs.bab.test": {Role: "home"}}
+ }
+ applyEdgesClaims(context.Background(), c, routes, served())
+ clock = clock.Add(edgesClaimTTL)
+ hang = true
+ hosts := served()
+ start := time.Now()
+ applyEdgesClaims(context.Background(), c, routes, hosts)
+ if took := time.Since(start); took > time.Second {
+ t.Fatalf("the pass took %s against a store that never answers, want about its 100ms deadline", took)
+ }
+ if h, ok := hosts["mine.bab.test"]; !ok || h.Stamp != 41 {
+ t.Fatalf("mine = %+v, %v: a read cut short by the deadline moved a served host", h, ok)
+ }
+ if _, ok := hosts["theirs.bab.test"]; ok {
+ t.Fatal("a read cut short by the deadline published a host this box was told not to")
+ }
+ if d := newEdgesClaims("annie", read).passDeadline; d != 20*time.Second {
+ t.Fatalf("the default pass deadline is %s, want 20s", d)
+ }
+}
+
+// Final review 4: a host hostclaim.Key rejects is never read; it publishes
+// as it did before claims.
+func TestEdgesClaimsNeverReadAHostKeyRejects(t *testing.T) {
+ const bad = "under_score.bab.test"
+ if _, err := hostclaim.Key(bad); err == nil {
+ t.Fatalf("hostclaim.Key accepts %q; the test needs a host it rejects", bad)
+ }
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{bad: serving("bert", 42)}}
+ c := newEdgesClaims("annie", book.read)
+ routes := []registry.Route{{Host: bad, Kind: "instance", Socket: "/run/carlos/x.sock", Provisioned: true}}
+ hosts := map[string]edgeHost{bad: {Role: "home"}}
+ applyEdgesClaims(context.Background(), c, routes, hosts)
+ if len(book.reads) != 0 {
+ t.Fatalf("read claims for %v", book.reads)
+ }
+ if h, ok := hosts[bad]; !ok || h.Stamp != 0 {
+ t.Fatalf("%s = %+v, %v; want home, unstamped, as before", bad, h, ok)
+ }
+}
+
+// Final review 4: claim read failures are logged on the first failing
+// pass and every 60th after it, and once when reads recover.
+func TestEdgesClaimsLogReadFailuresThrottled(t *testing.T) {
+ book := &claimsBook{claims: map[string]*hostclaim.Claim{}, errs: map[string]bool{"a.bab.test": true}}
+ 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 lines []string
+ c.logf = func(format string, a ...any) { lines = append(lines, fmt.Sprintf(format, a...)) }
+ for i := 0; i < 120; i++ {
+ c.decide(context.Background(), []string{"a.bab.test"})
+ clock = clock.Add(edgesClaimTTL)
+ }
+ if len(lines) != 3 { // passes 1, 60 and 120
+ t.Fatalf("%d lines for 120 failing passes, want 3: %q", len(lines), lines)
+ }
+ book.errs = nil
+ c.decide(context.Background(), []string{"a.bab.test"})
+ clock = clock.Add(edgesClaimTTL)
+ c.decide(context.Background(), []string{"a.bab.test"})
+ if len(lines) != 4 {
+ t.Fatalf("lines after recovery = %q, want one recovery line", lines)
+ }
+}