amadan

carlos / platform Public

edge: cert mirror and home map re-read only what moved; mesh keys keep their own poll (#336)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a0e3fb0a4c21f88055c0a79e0fb0a77b7c8ff9b0 parent a732043
6 files changed, +826 −42
  • internal/edge/certsync.go +228 −39
  • internal/edge/certsync_test.go +350 −0
  • internal/edge/edge.go +17 −0
  • internal/edge/homemap.go +88 −3
  • internal/edge/homemap_test.go +137 −0
  • internal/host/agent.go +6 −0
diff --git a/internal/edge/certsync.go b/internal/edge/certsync.go
index da88d527f..cbea09f81 100644
--- a/internal/edge/certsync.go
+++ b/internal/edge/certsync.go
@@ -36,6 +36,7 @@ import (
"errors"
"fmt"
"log"
+ "math/rand/v2"
"net"
"net/http"
"os"
@@ -521,6 +522,27 @@ type certMirror struct {
// latches (region HA spec 2 §6); nil without region HA.
health *lbhealth.State
+ // watch resolves the process's change-index watch (#336), handed down by
+ // the agent through WithChangeWatch and read on every poll: the agent
+ // arms its watch on its own schedule, possibly after the edge starts.
+ // nil, or a nil watch, is today's mirror: every poll a full Refresh.
+ // now is the poll's clock (nil is time.Now; tests inject one).
+ watch func() *release.ChangeWatch
+ now func() time.Time
+
+ // Guarded by refreshMu: what the published snapshot was read at.
+ // etags is each entry's ETag as its Get returned it ("" = unknown, read
+ // it again); lastFull is the last full rebuild that landed and fullAge
+ // the jittered backstop after which the next poll rebuilds in full;
+ // actedGen is the control/certs generation, captured BEFORE the read,
+ // that the snapshot is known to reflect (actedOK false = no such
+ // generation: the watch could not vouch, or some object went unread).
+ etags map[string]string
+ lastFull time.Time
+ fullAge time.Duration
+ actedGen uint64
+ actedOK bool
+
// synced remembers, per name, the mirrored bytes last reconciled into the
// local cache, so a steady state costs no local I/O at all.
syncMu sync.Mutex
@@ -552,6 +574,7 @@ func (e *Edge) newCertMirror(local autocert.Cache) *certMirror {
}
m := newCertMirror(e.staticStore, local)
m.health = e.lbHealth
+ m.watch = e.changeWatch
m.gated = func() bool {
_, held := e.issuerGate()
return !held
@@ -615,10 +638,12 @@ func newCertMirror(s release.Store, local autocert.Cache) *certMirror {
}
}
-// Refresh rebuilds the whole map from one list plus a get per key. No delta
-// cleverness: the prefix holds one small object per host and renewals are
-// rare, and a full rebuild is what makes a withdrawn entry (a spent challenge
-// token) actually leave.
+// Refresh rebuilds the whole map from one list plus a get per key. It is the
+// boot read before the listeners bind, every poll while the change index is
+// disarmed, and the armed loop's backstop and "unknown" read (poll, #336).
+// A rebuild from the listing is what makes a withdrawn entry (a spent
+// challenge token) actually leave; poll's changed-only read builds from a
+// listing too, for the same reason.
//
// An error leaves the last-good snapshot in place and is returned for the
// caller to log. That is the edge's standing degrade-not-die posture, and it
@@ -627,14 +652,23 @@ func newCertMirror(s release.Store, local autocert.Cache) *certMirror {
func (m *certMirror) Refresh(ctx context.Context) error {
m.refreshMu.Lock()
defer m.refreshMu.Unlock()
+ return m.refreshFull(ctx)
+}
+
+// refreshFull is Refresh's body, under refreshMu.
+func (m *certMirror) refreshFull(ctx context.Context) error {
+ // The generation this read will reflect, captured BEFORE the list: a
+ // write that lands during the read leaves control/certs moved past it,
+ // so the next poll looks again (#336). Without a usable watch there is
+ // none, and the next poll cannot skip.
+ gen, genOK := m.certsGen()
keys, err := m.s.List(ctx, certPrefix)
if err != nil {
return fmt.Errorf("list %s: %w", certPrefix, err)
}
prev := m.snapshot()
next := make(map[string][]byte, len(keys))
- var tried, failed int
- var lastErr error
+ etags := make(map[string]string, len(keys))
// The eligible names first, then their bodies CONCURRENTLY. The fetch is
// one round trip per object and the objects are independent, so doing
@@ -655,10 +689,7 @@ func (m *certMirror) Refresh(ctx context.Context) error {
// box pays count x latency, so each new published cert moves every box
// closer to the same cliff. us-east-1 at ~75ms is about a third of the
// way there at today's count.
- type fetch struct {
- key, name string
- }
- var items []fetch
+ var items []certFetch
for _, key := range keys {
name, ok := strings.CutPrefix(key, certPrefix)
// publishableCertName is the same filter the publish side applies:
@@ -667,15 +698,135 @@ func (m *certMirror) Refresh(ctx context.Context) error {
if !ok || !publishableCertName(name) {
continue
}
- items = append(items, fetch{key: key, name: name})
+ items = append(items, certFetch{key: key, name: name})
+ }
+ tried := len(items)
+ failed, lastErr := applyCertFetches(items, m.fetchCerts(ctx, items), prev, next, etags)
+ if tried > 0 && failed == tried {
+ // Nothing at all could be read: that is a bucket fault, not an object
+ // fault, and it is reported like a failed List — last-good stands.
+ return fmt.Errorf("get %s*: every one of %d object(s) failed, last: %w", certPrefix, tried, lastErr)
}
- tried = len(items)
+ if failed > 0 {
+ m.logPartialPoll(failed, tried, lastErr)
+ }
+ m.publish(ctx, prev, next, etags)
+ // A partial rebuild still counts as the full rebuild: the objects it
+ // could not read carry an unknown ETag, so the next poll's changed-only
+ // read fetches exactly those again. It reflects no generation, though,
+ // so that next poll cannot skip.
+ m.lastFull, m.fullAge = m.clock(), jitteredBackstop()
+ m.actedGen, m.actedOK = gen, genOK && failed == 0
+ return nil
+}
- type result struct {
- body []byte
- err error
+// poll is the 30s loop's refresh (#336). Disarmed, it is Refresh. Armed, it
+// polls the change index (a 304 when nothing moved, one GET either way) and
+// then does the least that keeps the snapshot current:
+//
+// - nothing, when control/certs has not moved since the generation the
+// snapshot reflects and the last full rebuild is younger than the
+// jittered backstop;
+// - a changed-only read (refreshChanged) when it has moved;
+// - today's full Refresh when the watch cannot vouch for the scope (the
+// index unreadable: "unknown", never "unchanged"), when no full rebuild
+// has landed yet, or at the backstop.
+//
+// Poll's moved list is not consulted: the watch is shared with the agent's
+// tick and record view, so whichever polls first consumes it. Generations
+// are compared with the one this mirror last acted on instead.
+func (m *certMirror) poll(ctx context.Context) error {
+ w := m.changeWatch()
+ if w == nil {
+ return m.Refresh(ctx)
+ }
+ // An error leaves the watch unusable, which Gen reports below; the poll
+ // itself has nothing to add.
+ _, _ = w.Poll(ctx)
+ m.refreshMu.Lock()
+ defer m.refreshMu.Unlock()
+ gen, ok := w.Gen(certScope)
+ switch {
+ case !ok, m.lastFull.IsZero(), !m.clock().Before(m.lastFull.Add(m.fullAge)):
+ return m.refreshFull(ctx)
+ case m.actedOK && gen == m.actedGen:
+ return nil
+ }
+ return m.refreshChanged(ctx, gen)
+}
+
+// refreshChanged rebuilds the snapshot from one ETag listing plus a get per
+// new or rewritten object, under refreshMu. gen is the control/certs
+// generation captured before the listing. It keeps every guarantee of the
+// full rebuild:
+//
+// - a key absent from the listing (a spent challenge token) is absent from
+// the new snapshot, because the snapshot is built from the listing;
+// - a failed get of a changed object carries the previous copy forward
+// with an unknown ETag, so the next poll fetches it again;
+// - the new snapshot is built whole and swapped in by publish, which also
+// copies changed material into the local cache and wakes the wildcard
+// manager, exactly as Refresh does.
+//
+// A store without the ETag listing, or a listing that fails, is "unknown":
+// today's full Refresh runs instead.
+func (m *certMirror) refreshChanged(ctx context.Context, gen uint64) error {
+ lister, ok := m.s.(release.ETagLister)
+ if !ok {
+ return m.refreshFull(ctx)
+ }
+ listed, err := lister.ListETags(ctx, certPrefix)
+ if err != nil {
+ return m.refreshFull(ctx)
+ }
+ prev := m.snapshot()
+ next := make(map[string][]byte, len(listed))
+ etags := make(map[string]string, len(listed))
+ var items []certFetch
+ for _, le := range listed {
+ name, ok := strings.CutPrefix(le.Key, certPrefix)
+ if !ok || !publishableCertName(name) {
+ continue
+ }
+ if was := m.etags[name]; was != "" && was == le.ETag {
+ if body, ok := prev.entries[name]; ok {
+ next[name], etags[name] = body, was
+ continue
+ }
+ }
+ items = append(items, certFetch{key: le.Key, name: name})
+ }
+ tried := len(items)
+ failed, lastErr := applyCertFetches(items, m.fetchCerts(ctx, items), prev, next, etags)
+ if tried > 0 && failed == tried {
+ // Refresh's rule: nothing that changed could be read, so the bucket
+ // is at fault and last-good stands whole. The generation stays
+ // unacted, so the next poll tries again.
+ return fmt.Errorf("get %s*: every one of %d changed object(s) failed, last: %w", certPrefix, tried, lastErr)
}
- results := make([]result, len(items))
+ if failed > 0 {
+ m.logPartialPoll(failed, tried, lastErr)
+ }
+ m.publish(ctx, prev, next, etags)
+ m.actedGen, m.actedOK = gen, failed == 0
+ return nil
+}
+
+// certFetch is one object a refresh reads; certFetchResult its outcome.
+type certFetch struct {
+ key, name string
+}
+
+type certFetchResult struct {
+ body []byte
+ etag string
+ err error
+}
+
+// fetchCerts gets every item's body, at most certRefreshConcurrency at once,
+// and returns the results in item order.
+func (m *certMirror) fetchCerts(ctx context.Context, items []certFetch) []certFetchResult {
+ results := make([]certFetchResult, len(items))
// A fixed pool pulling indices off a channel, NOT a goroutine per object
// parked on a semaphore. The distinction only shows up at scale, and it is
// the whole point of having a bound: goroutine-per-object costs one stack
@@ -696,8 +847,8 @@ func (m *certMirror) Refresh(ctx context.Context) error {
go func() {
defer wg.Done()
for i := range idx {
- body, _, err := m.s.Get(ctx, items[i].key)
- results[i] = result{body: body, err: err}
+ body, etag, err := m.s.Get(ctx, items[i].key)
+ results[i] = certFetchResult{body: body, etag: etag, err: err}
}
}()
}
@@ -706,57 +857,63 @@ func (m *certMirror) Refresh(ctx context.Context) error {
case idx <- i:
case <-ctx.Done():
// Leaves results[i:] zero — a nil body with a nil error, which
- // the apply loop below reads as a successful empty get. Set the
+ // the apply loop reads as a successful empty get. Set the
// context's error on the rest so they take the failed-get branch
// and carry the previous copy forward instead.
for j := i; j < len(items); j++ {
- results[j] = result{err: ctx.Err()}
+ results[j] = certFetchResult{err: ctx.Err()}
}
close(idx)
wg.Wait()
- goto apply
+ return results
}
}
close(idx)
wg.Wait()
-apply:
+ return results
+}
+// applyCertFetches folds fetched bodies into next (and their ETags into
+// etags), carrying prev's copy of any object that failed, and returns the
+// failure count and the last failure.
+func applyCertFetches(items []certFetch, results []certFetchResult, prev *certSnapshot, next map[string][]byte, etags map[string]string) (failed int, lastErr error) {
// Applied in key order, not completion order, so a partial poll reports
// the same lastErr whichever way the fetches happened to interleave.
for i, it := range items {
- body, err := results[i].body, results[i].err
+ r := results[i]
switch {
- case errors.Is(err, os.ErrNotExist):
+ case errors.Is(r.err, os.ErrNotExist):
// Raced a delete between the list and the get — a spent token. Not
// a fault; the entry is simply gone.
continue
- case err != nil:
+ case r.err != nil:
// ONE bad object is not a bad poll. Aborting here would discard the
// whole rebuild — every unrelated cert that landed this tick with
// it — and skip syncLocal and the wildcard wake; a single
// permanently unreadable key would freeze the mirror forever. Carry
- // the previous copy of this name forward and keep going.
+ // the previous copy of this name forward and keep going, with an
+ // unknown ETag so the next changed-only read fetches it again.
failed++
- lastErr = err
+ lastErr = r.err
if body, ok := prev.entries[it.name]; ok {
- next[it.name] = body
+ next[it.name], etags[it.name] = body, ""
}
continue
}
- next[it.name] = body
- }
- if tried > 0 && failed == tried {
- // Nothing at all could be read: that is a bucket fault, not an object
- // fault, and it is reported like a failed List — last-good stands.
- return fmt.Errorf("get %s*: every one of %d object(s) failed, last: %w", certPrefix, tried, lastErr)
- }
- if failed > 0 {
- m.logPartialPoll(failed, tried, lastErr)
+ next[it.name], etags[it.name] = r.body, r.etag
}
+ return failed, lastErr
+}
+
+// publish swaps in the snapshot built from next, whole, and does what every
+// landed refresh does after it: the local cache copy-in, the health latch and
+// the wildcard wake. Under refreshMu.
+func (m *certMirror) publish(ctx context.Context, prev *certSnapshot, next map[string][]byte, etags map[string]string) {
snap := buildCertSnapshot(prev, next)
m.mu.Lock()
m.snap = snap
m.mu.Unlock()
+ m.etags = etags
m.syncLocal(ctx, snap)
// The local cache the handshake reads now holds what the bucket does:
// the certificate condition is met for good (§6). A later failed
@@ -767,7 +924,39 @@ apply:
// serving path in one poll instead of up to twelve hours.
m.wake()
}
- return nil
+}
+
+// certScope is the change-index scope every write under certPrefix bumps.
+const certScope = "control/certs"
+
+func (m *certMirror) changeWatch() *release.ChangeWatch {
+ if m.watch == nil {
+ return nil
+ }
+ return m.watch()
+}
+
+// certsGen is control/certs' generation as the watch last saw it, ok false
+// when there is no usable watch.
+func (m *certMirror) certsGen() (uint64, bool) {
+ if w := m.changeWatch(); w != nil {
+ return w.Gen(certScope)
+ }
+ return 0, false
+}
+
+func (m *certMirror) clock() time.Time {
+ if m.now != nil {
+ return m.now()
+ }
+ return time.Now()
+}
+
+// jitteredBackstop is release.ChangeIndexBackstop jittered by up to 10% either
+// way (R3), drawn per full read so a fleet's rebuilds do not line up.
+func jitteredBackstop() time.Duration {
+ b := release.ChangeIndexBackstop
+ return b - b/10 + time.Duration(rand.Int64N(int64(b/5)+1))
}
// logPartialPoll reports unreadable objects on the loop's own first-and-every-
@@ -908,7 +1097,7 @@ func (m *certMirror) run(stop <-chan struct{}) {
case <-tick.C:
}
ctx, cancel := context.WithTimeout(context.Background(), certMirrorTimeout)
- err := m.Refresh(ctx)
+ err := m.poll(ctx)
cancel()
if err != nil {
errs++
diff --git a/internal/edge/certsync_test.go b/internal/edge/certsync_test.go
index 0dd51fc51..681bed511 100644
--- a/internal/edge/certsync_test.go
+++ b/internal/edge/certsync_test.go
@@ -14,6 +14,7 @@ import (
"runtime"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -1951,3 +1952,352 @@ func TestCertMirrorRefreshFetchesConcurrently(t *testing.T) {
}
}
}
+
+// ---- #336: the change index gates the mirror's reads -------------------------
+
+// changeIndexStore is a DirStore with the change index on that records every
+// read by method and key. Embedding the concrete store forwards
+// ConditionalGetter and ETagLister, so the mirror and the watch see the store
+// production hands them, and every write bumps its scope as in production.
+type changeIndexStore struct {
+ *release.DirStore
+ mu sync.Mutex
+ calls []string
+ failGet string // Get of exactly this key fails
+}
+
+func (s *changeIndexStore) setFailGet(key string) { s.mu.Lock(); s.failGet = key; s.mu.Unlock() }
+
+func newChangeIndexStore(t *testing.T) *changeIndexStore {
+ t.Helper()
+ d := release.NewDirStore(t.TempDir())
+ d.EnableChangeIndex()
+ return &changeIndexStore{DirStore: d}
+}
+
+func (s *changeIndexStore) note(method, key string) {
+ s.mu.Lock()
+ s.calls = append(s.calls, method+" "+key)
+ s.mu.Unlock()
+}
+
+func (s *changeIndexStore) reset() { s.mu.Lock(); s.calls = nil; s.mu.Unlock() }
+
+// count is how many recorded calls of method name a key under prefix.
+func (s *changeIndexStore) count(method, prefix string) int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ n := 0
+ for _, c := range s.calls {
+ if strings.HasPrefix(c, method+" "+prefix) {
+ n++
+ }
+ }
+ return n
+}
+
+func (s *changeIndexStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.note("Get", key)
+ s.mu.Lock()
+ fail := s.failGet != "" && s.failGet == key
+ s.mu.Unlock()
+ if fail {
+ return nil, "", errors.New("get refused")
+ }
+ return s.DirStore.Get(ctx, key)
+}
+
+func (s *changeIndexStore) GetIfChanged(ctx context.Context, key, etag string) ([]byte, string, bool, error) {
+ s.note("GetIfChanged", key)
+ return s.DirStore.GetIfChanged(ctx, key, etag)
+}
+
+func (s *changeIndexStore) List(ctx context.Context, prefix string) ([]string, error) {
+ s.note("List", prefix)
+ return s.DirStore.List(ctx, prefix)
+}
+
+func (s *changeIndexStore) ListETags(ctx context.Context, prefix string) ([]release.KeyETag, error) {
+ s.note("ListETags", prefix)
+ return s.DirStore.ListETags(ctx, prefix)
+}
+
+// certReads is every bucket read the mirror made under control/certs/.
+func certReads(s *changeIndexStore) (list, listETags, get int) {
+ return s.count("List", certPrefix), s.count("ListETags", certPrefix), s.count("Get", certPrefix)
+}
+
+// armedMirror is a mirror over s with the change index armed: one watch over
+// the same store (the agent's, in production) and an injected clock.
+func armedMirror(t *testing.T, s *changeIndexStore) (*certMirror, *testClock) {
+ t.Helper()
+ w := release.NewChangeWatch(s)
+ clk := &testClock{t: time.Date(2026, 10, 21, 12, 0, 0, 0, time.UTC)}
+ m := newCertMirror(s, nil)
+ m.watch = func() *release.ChangeWatch { return w }
+ m.now = clk.now
+ return m, clk
+}
+
+func mirrorPoll(t *testing.T, m *certMirror) {
+ t.Helper()
+ if err := m.poll(context.Background()); err != nil {
+ t.Fatalf("poll: %v", err)
+ }
+}
+
+func TestCertMirrorSkipsWhenUnmoved(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ putBucket(t, s, "shop.example.com", []byte("PEM2"))
+ m, clk := armedMirror(t, s)
+ mirrorPoll(t, m)
+ if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
+ t.Fatalf("first poll: mirror cert = %q, %v", b, ok)
+ }
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l+le+g != 0 {
+ t.Fatalf("unmoved poll read the bucket: List %d, ListETags %d, Get %d; want none", l, le, g)
+ }
+ if n := s.count("GetIfChanged", release.ChangeIndexKey); n != 1 {
+ t.Fatalf("unmoved poll: %d index reads, want 1", n)
+ }
+ if b, ok := m.Cert("shop.example.com"); !ok || string(b) != "PEM2" {
+ t.Fatalf("after a skipped poll the mirror = %q, %v — want the same copy", b, ok)
+ }
+}
+
+func TestCertMirrorFetchesOnlyChanged(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ local := newCountingCache(t)
+ pemA := hostCertPEM(t, "a.example.com", time.Now().Add(60*24*time.Hour))
+ pemB := hostCertPEM(t, "b.example.com", time.Now().Add(60*24*time.Hour))
+ putBucket(t, s, "a.example.com", pemA)
+ putBucket(t, s, "b.example.com", pemB)
+ putBucket(t, s, "c.example.com", []byte("PEM-C"))
+ m, clk := armedMirror(t, s)
+ m.local = local
+ mirrorPoll(t, m)
+
+ renewed := hostCertPEM(t, "b.example.com", time.Now().Add(90*24*time.Hour))
+ putBucket(t, s, "b.example.com", renewed)
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ l, le, g := certReads(s)
+ if l != 0 || le != 1 || g != 1 {
+ t.Fatalf("one rewritten cert: List %d, ListETags %d, Get %d; want 0, 1, 1", l, le, g)
+ }
+ if n := s.count("Get", certKey("b.example.com")); n != 1 {
+ t.Fatalf("the one Get must be the rewritten cert's, got %d of it", n)
+ }
+ if b, ok := m.Cert("b.example.com"); !ok || !bytes.Equal(b, renewed) {
+ t.Fatal("the rewritten cert must be served from the new snapshot")
+ }
+ for _, name := range []string{"a.example.com", "c.example.com"} {
+ if _, ok := m.Cert(name); !ok {
+ t.Fatalf("%s did not move and must stay in the snapshot", name)
+ }
+ }
+ if got, err := localCert(t, local, "b.example.com"); err != nil || !bytes.Equal(got, renewed) {
+ t.Fatalf("a changed cert must still be copied into the local cache: %v", err)
+ }
+ if m.snapshot().certs["b.example.com"] == nil {
+ t.Fatal("the changed cert must be parsed into the new snapshot")
+ }
+}
+
+func TestCertMirrorDropsRemovedKey(t *testing.T) {
+ t.Parallel()
+ const token = "0123456789abcdefghij0123456789abcdefghijkl"
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ putBucket(t, s, token+httpTokenSuffix, []byte("resp"))
+ m, clk := armedMirror(t, s)
+ mirrorPoll(t, m)
+ if _, ok := m.Challenge(token); !ok {
+ t.Fatal("precondition: the live token is mirrored")
+ }
+ if err := s.Delete(context.Background(), certKey(token+httpTokenSuffix)); err != nil {
+ t.Fatal(err)
+ }
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if _, ok := m.Challenge(token); ok {
+ t.Fatal("a spent challenge token must leave the snapshot when its key leaves the listing")
+ }
+ if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
+ t.Fatalf("the unmoved cert = %q, %v — want it kept", b, ok)
+ }
+ if l, le, g := certReads(s); l != 0 || le != 1 || g != 0 {
+ t.Fatalf("a deletion: List %d, ListETags %d, Get %d; want 0, 1, 0", l, le, g)
+ }
+}
+
+func TestCertMirrorFullRebuildAtBackstop(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ putBucket(t, s, "shop.example.com", []byte("PEM2"))
+ m, clk := armedMirror(t, s)
+ mirrorPoll(t, m)
+
+ // Inside the jitter's lower bound (-10%): still trusted.
+ s.reset()
+ clk.advance(release.ChangeIndexBackstop*9/10 - time.Minute)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l+le+g != 0 {
+ t.Fatalf("a young snapshot with the index unmoved: List %d, ListETags %d, Get %d; want none", l, le, g)
+ }
+
+ // Past the jitter's upper bound (+10%): a full rebuild, index unmoved.
+ s.reset()
+ clk.advance(release.ChangeIndexBackstop*2/10 + 2*time.Minute)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l != 1 || le != 0 || g != 2 {
+ t.Fatalf("at the backstop: List %d, ListETags %d, Get %d; want the full rebuild 1, 0, 2", l, le, g)
+ }
+
+ // The rebuild resets the age: the next unmoved poll reads nothing.
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l+le+g != 0 {
+ t.Fatalf("after the backstop rebuild: List %d, ListETags %d, Get %d; want none", l, le, g)
+ }
+}
+
+func TestCertMirrorBootRefreshUnchanged(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ putBucket(t, s, "shop.example.com", []byte("PEM2"))
+ m, clk := armedMirror(t, s)
+ // The agent's tick has already polled the shared watch.
+ if _, err := m.watch().Poll(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ s.reset()
+ // The boot path: a full read, whatever the index says, and twice over.
+ for i := 0; i < 2; i++ {
+ if err := m.Refresh(context.Background()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if l, le, g := certReads(s); l != 2 || le != 0 || g != 4 {
+ t.Fatalf("two boot refreshes: List %d, ListETags %d, Get %d; want 2, 0, 4", l, le, g)
+ }
+ if n := s.count("GetIfChanged", ""); n != 0 {
+ t.Fatalf("the boot refresh must not poll the index (%d reads): it runs before the listeners bind", n)
+ }
+ // The loop's first poll trusts the boot read.
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l+le+g != 0 {
+ t.Fatalf("first poll after boot, index unmoved: List %d, ListETags %d, Get %d; want none", l, le, g)
+ }
+}
+
+// Disarmed (no watch handed down, or the agent never armed one), every poll
+// is today's full Refresh.
+func TestCertMirrorDisarmedPollIsAFullRefresh(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ for _, watch := range []func() *release.ChangeWatch{nil, func() *release.ChangeWatch { return nil }} {
+ m := newCertMirror(s, nil)
+ m.watch = watch
+ for i := 0; i < 2; i++ {
+ s.reset()
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l != 1 || le != 0 || g != 1 {
+ t.Fatalf("disarmed poll %d: List %d, ListETags %d, Get %d; want 1, 0, 1", i, l, le, g)
+ }
+ if n := s.count("GetIfChanged", ""); n != 0 {
+ t.Fatalf("disarmed poll read the index %d time(s)", n)
+ }
+ }
+ }
+}
+
+// A changed cert whose Get fails keeps its previous copy (degrade, don't
+// die), and the next poll reads it again with the index unmoved.
+func TestCertMirrorFailedGetOfAChangedCertKeepsThePreviousCopy(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putBucket(t, s, "example.com", []byte("PEM"))
+ putBucket(t, s, "shop.example.com", []byte("PEM2"))
+ m, clk := armedMirror(t, s)
+ mirrorPoll(t, m)
+
+ putBucket(t, s, "example.com", []byte("PEM-RENEWED"))
+ putBucket(t, s, "new.example.com", []byte("PEM3"))
+ s.setFailGet(certKey("example.com"))
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM" {
+ t.Fatalf("a failed Get of a changed cert: mirror = %q, %v — want the previous copy", b, ok)
+ }
+ if b, ok := m.Cert("new.example.com"); !ok || string(b) != "PEM3" {
+ t.Fatalf("the other changed cert = %q, %v — want it landed", b, ok)
+ }
+
+ s.setFailGet("")
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if b, ok := m.Cert("example.com"); !ok || string(b) != "PEM-RENEWED" {
+ t.Fatalf("the retry: mirror = %q, %v — want the renewed copy", b, ok)
+ }
+ if l, le, g := certReads(s); l != 0 || le != 1 || g != 1 {
+ t.Fatalf("the retry: List %d, ListETags %d, Get %d; want 0, 1, 1", l, le, g)
+ }
+ s.reset()
+ clk.advance(certMirrorInterval)
+ mirrorPoll(t, m)
+ if l, le, g := certReads(s); l+le+g != 0 {
+ t.Fatalf("settled: List %d, ListETags %d, Get %d; want none", l, le, g)
+ }
+}
+
+// The scopes the mirror and the home map gate on are the ones their prefixes'
+// writes bump.
+func TestCertMirrorAndHomeMapScopesMatchTheirPrefixes(t *testing.T) {
+ t.Parallel()
+ if got, ok := release.ChangeScope(certKey("example.com")); !ok || got != certScope {
+ t.Fatalf("a cert write bumps %q (%v), the mirror gates on %q", got, ok, certScope)
+ }
+ if got, ok := release.ChangeScope(edgesPrefix + "box.json"); !ok || got != edgesScope {
+ t.Fatalf("an edges-file write bumps %q (%v), the home map gates on %q", got, ok, edgesScope)
+ }
+}
+
+// WithChangeWatch reaches both loops, resolved per poll: a watch the agent
+// arms after the edge started is still seen.
+func TestCertMirrorWithChangeWatchReachesTheMirrorAndHomeMap(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ var armed atomic.Pointer[release.ChangeWatch]
+ e, err := New(openReg(t), nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(e.probers.Stop)
+ e.staticStore = s
+ WithChangeWatch(armed.Load)(e)
+ m, hm := e.newCertMirror(nil), e.newHomeMap()
+ if m.changeWatch() != nil || hm.changeWatch() != nil {
+ t.Fatal("nothing armed yet: both loops must be disarmed")
+ }
+ w := release.NewChangeWatch(s)
+ armed.Store(w)
+ if m.changeWatch() != w || hm.changeWatch() != w {
+ t.Fatal("a watch armed after the edge started must reach both loops")
+ }
+}
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 31ea8b8de..a23121947 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -258,6 +258,10 @@ type Edge struct {
// this one and never the reverse. It is what the home map excludes its own
// hosts by; "" (bare `carlos edge`) leaves the map offering nothing.
box string
+ // changeWatch resolves the process's change-index watch (#336), handed
+ // down through WithChangeWatch. The cert mirror and the home map call it
+ // on every poll. nil, or a nil watch, keeps both on today's full reads.
+ changeWatch func() *release.ChangeWatch
// homes, when non-nil, is the mirror of control/edges/ plus the mesh keys
// (homemap.go): who homes a host this box does not serve, and what proves
// an inbound hop. Atomic because it is armed in RunWithActivatorReady after
@@ -326,6 +330,19 @@ func WithIssuerContender(c func() bool) RunOption { return func(e *Edge) { e.con
// See homemap.go.
func WithBoxLabel(label string) RunOption { return func(e *Edge) { e.box = label } }
+// WithChangeWatch hands the edge the agent's change-index watch (#336) as a
+// function, called on every cert-mirror and home-map poll. A function and
+// not the watch itself, because the agent may arm its watch after the edge
+// has started (armChangeIndex runs on a goroutine on some boxes), and an
+// edge holding a nil captured at start would stay disarmed for the life of
+// the process. One watch per process, the agent's: the edge never arms one
+// of its own, so CARLOS_CHANGE_INDEX and the fleet-box exclusion stay
+// decided in one place. The function, or the watch it returns, nil keeps
+// both loops on today's full reads.
+func WithChangeWatch(w func() *release.ChangeWatch) RunOption {
+ return func(e *Edge) { e.changeWatch = w }
+}
+
// WithEdgeDoor asks for the edge-local door (door.go, #175 P7) on the unix
// socket at path, presenting publicAddr, the box's public address, as the
// address of every door request. The door opens once the home map and the
diff --git a/internal/edge/homemap.go b/internal/edge/homemap.go
index f91901da4..e266bc40e 100644
--- a/internal/edge/homemap.go
+++ b/internal/edge/homemap.go
@@ -287,6 +287,24 @@ type homeMap struct {
// nil without region HA: Refresh latches the first loaded map and key
// and stamps every attempt it completes.
health *lbhealth.State
+
+ // watch and now are certMirror's (#336): the process's change-index
+ // watch, resolved on every poll (nil, or a nil watch, is today's map),
+ // and the poll's clock (nil is time.Now). Only the hosts half is gated
+ // on them; the mesh keys keep their own read on every poll.
+ watch func() *release.ChangeWatch
+ now func() time.Time
+
+ // Guarded by refreshMu: the last hosts rebuild that read every edges
+ // file (lastHosts), the jittered backstop after which the next poll
+ // reads them all again (hostsAge), and the control/edges generation,
+ // captured BEFORE that read, which the fold is known to reflect
+ // (actedOK false = none: the watch could not vouch, or a file went
+ // unread or unparsed and must be retried).
+ lastHosts time.Time
+ hostsAge time.Duration
+ actedGen uint64
+ actedOK bool
}
func newHomeMap(s release.Store, self string) *homeMap {
@@ -305,6 +323,7 @@ func (e *Edge) newHomeMap() *homeMap {
m := newHomeMap(e.staticStore, e.box)
m.privateDial = regionHAFromEnv(os.Getenv)
m.health = e.lbHealth
+ m.watch = e.changeWatch
return m
}
@@ -320,17 +339,72 @@ func (e *Edge) newHomeMap() *homeMap {
func (m *homeMap) Refresh(ctx context.Context) error {
m.refreshMu.Lock()
defer m.refreshMu.Unlock()
- hostsErr, keysErr := m.refreshHosts(ctx), m.refreshKeys(ctx)
+ return m.refresh(ctx, true)
+}
+
+// poll is the 30s loop's refresh (#336). Disarmed, it is Refresh. Armed, it
+// polls the change index and skips the hosts half when control/edges has not
+// moved since the generation the fold reflects and the last full hosts read
+// is younger than the jittered backstop; otherwise (moved, the watch unable
+// to vouch, no read yet, or at the backstop) the hosts half runs as today.
+// The mesh keys are read on every poll whatever the index says: their
+// objects (control/edge-mesh-key and .next) bump no scope, so a rotation
+// must not wait on control/edges (codex 8). Poll's moved list is advisory
+// (the watch is shared), so the decision compares generations instead.
+func (m *homeMap) poll(ctx context.Context) error {
+ w := m.changeWatch()
+ if w == nil {
+ return m.Refresh(ctx)
+ }
+ // An error leaves the watch unusable, which Gen reports below.
+ _, _ = w.Poll(ctx)
+ m.refreshMu.Lock()
+ defer m.refreshMu.Unlock()
+ gen, ok := w.Gen(edgesScope)
+ current := ok && m.actedOK && gen == m.actedGen &&
+ !m.lastHosts.IsZero() && m.clock().Before(m.lastHosts.Add(m.hostsAge))
+ return m.refresh(ctx, !current)
+}
+
+// refresh is Refresh's body, under refreshMu. readHosts false skips the
+// hosts half, which then counts as a success: the fold it keeps is one a
+// clean read landed (poll skips only after one, and that read latched
+// MapLoaded), so the latch below is a no-op and the health answer is what a
+// full read would have given. The attempt is stamped either way, so the
+// five-minute refresh heartbeat holds across skipped polls.
+func (m *homeMap) refresh(ctx context.Context, readHosts bool) error {
+ var hostsErr error
+ if readHosts {
+ hostsErr = m.refreshHosts(ctx)
+ }
+ keysErr := m.refreshKeys(ctx)
if hostsErr == nil {
m.health.MapLoaded()
}
if current, _ := m.meshKeys(); keysErr == nil && len(current) > 0 {
m.health.KeyLoaded()
}
- m.health.RefreshAttempted(time.Now())
+ m.health.RefreshAttempted(m.clock())
return errors.Join(hostsErr, keysErr)
}
+// edgesScope is the change-index scope every write under edgesPrefix bumps.
+const edgesScope = "control/edges"
+
+func (m *homeMap) changeWatch() *release.ChangeWatch {
+ if m.watch == nil {
+ return nil
+ }
+ return m.watch()
+}
+
+func (m *homeMap) clock() time.Time {
+ if m.now != nil {
+ return m.now()
+ }
+ return time.Now()
+}
+
// refreshHosts rebuilds the whole map from one list plus a get per key — no
// delta cleverness, for certMirror.Refresh's reason: a full rebuild is what
// makes a RETIRED box's hosts actually leave the map.
@@ -340,6 +414,13 @@ func (m *homeMap) Refresh(ctx context.Context) error {
// backhaul in the deployment at once, turning a blip into a regional outage
// for every host homed elsewhere.
func (m *homeMap) refreshHosts(ctx context.Context) error {
+ // The generation this read will reflect, captured BEFORE the list, so a
+ // write landing during the read leaves control/edges moved past it (#336).
+ var gen uint64
+ var genOK bool
+ if w := m.changeWatch(); w != nil {
+ gen, genOK = w.Gen(edgesScope)
+ }
keys, err := m.s.List(ctx, edgesPrefix)
if err != nil {
return fmt.Errorf("list %s: %w", edgesPrefix, err)
@@ -407,6 +488,10 @@ func (m *homeMap) refreshHosts(ctx context.Context) error {
m.snap = snap
wake := m.hostsWake
m.mu.Unlock()
+ // A partial read reflects no generation: the file it could not read or
+ // parse is retried on the next poll, as it always was.
+ m.lastHosts, m.hostsAge = m.clock(), jitteredBackstop()
+ m.actedGen, m.actedOK = gen, genOK && failed == 0
if wake != nil && !samePlatformSet(prev.hosts, snap.hosts) {
// The deployment's platform name set moved: a wildcard parent may
// have appeared for an account homed elsewhere (#300). Off this
@@ -922,7 +1007,7 @@ func (m *homeMap) run(stop <-chan struct{}) {
case <-tick.C:
}
ctx, cancel := context.WithTimeout(context.Background(), homeMapTimeout)
- err := m.Refresh(ctx)
+ err := m.poll(ctx)
cancel()
if err != nil {
errs++
diff --git a/internal/edge/homemap_test.go b/internal/edge/homemap_test.go
index cdf518660..9c3e786ea 100644
--- a/internal/edge/homemap_test.go
+++ b/internal/edge/homemap_test.go
@@ -18,6 +18,7 @@ import (
"time"
"github.com/carlosframework/platform/internal/edgeproof"
+ "github.com/carlosframework/platform/internal/lbhealth"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -1784,3 +1785,139 @@ func TestLocalWildcardYieldsToARemoteExactHost(t *testing.T) {
t.Fatalf("home edge saw %d request(s), want still 1", n)
}
}
+
+// ---- #336: only the hosts half is gated on the change index ------------------
+
+// armedHomeMap is a home map over s with the change index armed (one watch over
+// the same store, the agent's in production) and a clock that starts now, so
+// the health state's real-time reads agree with it.
+func armedHomeMap(t *testing.T, s *changeIndexStore, self string) (*homeMap, *testClock) {
+ t.Helper()
+ w := release.NewChangeWatch(s)
+ clk := &testClock{t: time.Now()}
+ m := newHomeMap(s, self)
+ m.watch = func() *release.ChangeWatch { return w }
+ m.now = clk.now
+ return m, clk
+}
+
+func homePoll(t *testing.T, m *homeMap) {
+ t.Helper()
+ if err := m.poll(context.Background()); err != nil {
+ t.Fatalf("poll: %v", err)
+ }
+}
+
+func edgesReads(s *changeIndexStore) int {
+ return s.count("List", edgesPrefix) + s.count("ListETags", edgesPrefix) + s.count("Get", edgesPrefix)
+}
+
+// codex 8: a mesh-key rotation writes control/edge-mesh-key, which bumps no
+// scope, so the keys must not wait on control/edges.
+func TestHomeMapKeysPolledIndependently(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
+ putJSON(t, s, meshKeyKey, strings.Repeat("aa", 32))
+ m, clk := armedHomeMap(t, s, "annie")
+ homePoll(t, m)
+ if _, _, ok := m.Home("b.test"); !ok {
+ t.Fatal("precondition: the first poll folds the edges files")
+ }
+
+ putJSON(t, s, meshKeyKey, strings.Repeat("bb", 32))
+ s.reset()
+ clk.advance(homeMapInterval)
+ homePoll(t, m)
+ if cur, _ := m.meshKeys(); string(cur) != strings.Repeat("\xbb", 32) {
+ t.Fatalf("the rotated mesh key must load within one poll, got %x", cur)
+ }
+ if n := edgesReads(s); n != 0 {
+ t.Fatalf("control/edges did not move, yet the poll made %d read(s) under it", n)
+ }
+ if n := s.count("Get", meshKeyKey); n < 1 {
+ t.Fatal("the mesh key must be read on every poll")
+ }
+ if _, _, ok := m.Home("b.test"); !ok {
+ t.Fatal("a skipped hosts read must keep the fold")
+ }
+}
+
+// codex 8: the five-minute refresh heartbeat holds across polls whose hosts
+// read is skipped, and the forwarding latch stays set.
+func TestHomeMapHeartbeatOnUnchangedPoll(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
+ putJSON(t, s, meshKeyKey, strings.Repeat("aa", 32))
+ m, clk := armedHomeMap(t, s, "annie")
+ health := &lbhealth.State{}
+ m.health = health
+ homePoll(t, m)
+ for i := 0; i < 4; i++ {
+ s.reset()
+ clk.advance(lbhealth.RefreshWindow - time.Minute)
+ homePoll(t, m)
+ if n := edgesReads(s); n != 0 {
+ t.Fatalf("unchanged poll %d made %d read(s) under control/edges", i, n)
+ }
+ for _, c := range health.Failing(clk.now()) {
+ if c == lbhealth.CondRefresh || c == lbhealth.CondForwarding {
+ t.Fatalf("unchanged poll %d: health fails %q — a skipped hosts read must still beat", i, c)
+ }
+ }
+ }
+}
+
+// The hosts half re-reads when control/edges moves, and at the backstop with
+// the index unmoved.
+func TestHomeMapHostsReadOnMoveAndAtBackstop(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
+ m, clk := armedHomeMap(t, s, "annie")
+ homePoll(t, m)
+
+ putJSON(t, s, "control/edges/carol.json", `{"v":1,"box":"carol","eip":"3.3.3.3","hosts":{"c.test":{"role":"home"}}}`)
+ s.reset()
+ clk.advance(homeMapInterval)
+ homePoll(t, m)
+ if _, _, ok := m.Home("c.test"); !ok {
+ t.Fatal("a moved control/edges must be read on the next poll")
+ }
+ if n := s.count("List", edgesPrefix); n != 1 {
+ t.Fatalf("moved: %d list(s) of control/edges, want 1", n)
+ }
+
+ s.reset()
+ clk.advance(release.ChangeIndexBackstop*9/10 - time.Minute)
+ homePoll(t, m)
+ if n := edgesReads(s); n != 0 {
+ t.Fatalf("young and unmoved: %d read(s), want none", n)
+ }
+ s.reset()
+ clk.advance(release.ChangeIndexBackstop*2/10 + 2*time.Minute)
+ homePoll(t, m)
+ if n := s.count("List", edgesPrefix); n != 1 {
+ t.Fatalf("at the backstop: %d list(s) of control/edges, want 1", n)
+ }
+}
+
+// Disarmed, every poll reads the hosts as today.
+func TestHomeMapDisarmedPollReadsHosts(t *testing.T) {
+ t.Parallel()
+ s := newChangeIndexStore(t)
+ putJSON(t, s, "control/edges/betty.json", `{"v":1,"box":"betty","eip":"2.2.2.2","hosts":{"b.test":{"role":"home"}}}`)
+ m := newHomeMap(s, "annie")
+ m.watch = func() *release.ChangeWatch { return nil }
+ for i := 0; i < 2; i++ {
+ s.reset()
+ homePoll(t, m)
+ if n := s.count("List", edgesPrefix); n != 1 {
+ t.Fatalf("disarmed poll %d: %d list(s), want 1", i, n)
+ }
+ if n := s.count("GetIfChanged", ""); n != 0 {
+ t.Fatalf("disarmed poll %d read the index", i)
+ }
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 2385e2cc8..1bee56397 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -899,6 +899,12 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// the two derivations must be one derivation. Without it the edge
// would offer nothing (fail-closed), never dial itself.
edge.WithBoxLabel(metricsBoxLabel(os.Getenv, readMachineID())),
+ // The change index (#336): the cert mirror and the home map read
+ // through the process's one watch. A loader, not the watch: on a box
+ // without claim writes the index is armed on runDuties' goroutine,
+ // which may not have run yet. Disarmed, it returns nil for good and
+ // both keep today's full reads.
+ edge.WithChangeWatch(armedChangeWatch.Load),
},
// The edge-local door (#175 P7), only when CARLOS_EDGE_DOOR names a
// socket. It presents this box's public address, the one the edges