amadan

carlos / platform Public

edge: the wildcard manager wakes when a remote platform hold expires or revives; the wake compares the live holds at each poll's time

The home map's wake compared the snapshots' platform hold candidates, which
include expired holds: an expiry changed nothing it compared, and a hold
revived by a fresh fetch of the same file looked unchanged too, so the
parent left or returned only on an unrelated wake (Task 2 review Minor 2).
Refresh now ranks the platform holds at its own time on every poll, failed
ones included, and wakes when that live set differs from the last poll's.
Hosts derives from the same live set at its own now.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude 29097d9e0acc0e6513c3dc6dc657a10738231a7e parent 7772738
2 files changed, +219 −19
  • internal/edge/homemap.go +58 −19
  • internal/edge/homemap_test.go +161 −0
diff --git a/internal/edge/homemap.go b/internal/edge/homemap.go
index 61df9cc85..1eb3bc750 100644
--- a/internal/edge/homemap.go
+++ b/internal/edge/homemap.go
@@ -254,8 +254,9 @@ type homeSnapshot struct {
hosts map[string]homeEntry
// cands is every admitted candidate by name (buildCandidates), the
// input of the ranking; platformHolds the names a remote real holding
- // candidate marks platform, the second half of the wildcard parent
- // inputs (samePlatformSet's wake).
+ // candidate marks platform: the names that MAY be wildcard parent
+ // inputs. Whether one is depends on the time (a hold expires), so it
+ // is ranked at each evaluation (livePlatformHolds), never here.
cands map[string][]candidate
platformHolds map[string]bool
}
@@ -279,6 +280,10 @@ type homeMap struct {
snap *homeSnapshot
// hostsWake fires after a Refresh that changed the NAME set (setHostsWake).
hostsWake func()
+ // liveHolds is the live platform hold set (livePlatformHolds) the last
+ // Refresh evaluated, the wake's baseline for the half of the parent
+ // inputs that moves with the clock.
+ liveHolds map[string]bool
// partialPolls counts polls that could not read (or parse) every object,
// for the first-and-every-tenth log cadence.
partialPolls int
@@ -362,7 +367,10 @@ 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)
+ prev := m.snapshot()
+ hostsErr := m.refreshHosts(ctx)
+ m.wakeOnMove(prev, m.clock())
+ keysErr := m.refreshKeys(ctx)
if hostsErr == nil {
m.health.MapLoaded()
}
@@ -450,23 +458,58 @@ func (m *homeMap) refreshHosts(ctx context.Context) error {
snap := &homeSnapshot{files: next, hosts: hosts, cands: cands, platformHolds: platformHolds(cands)}
m.mu.Lock()
m.snap = snap
+ m.mu.Unlock()
+ return nil
+}
+
+// wakeOnMove fires the wake when the deployment's platform name set moved
+// between prev and the current snapshot: a wildcard parent may have appeared
+// for an account homed elsewhere (#300), or gone. The set is the ready
+// fold's platform names (samePlatformSet) plus the platform holds that are
+// LIVE at now (livePlatformHolds), compared with the live set the previous
+// Refresh evaluated rather than with the previous snapshot's candidates.
+//
+// Comparing live sets is what makes the clock wake the manager (self-serve
+// moves plan 3, Task 5): a remote hold expires with no fetch at all, and an
+// expired hold revived by a fresh fetch of the same file leaves the
+// candidates unchanged, so neither shows in the snapshots. Refresh calls this
+// on every poll, the failed ones included (a failed List or every Get
+// failing is exactly when files stop being re-fetched and their holds run
+// out), so the poll's own 30 s cadence is the time-based re-evaluation: an
+// expiry reaches the parent set within one interval, with no timer of its
+// own. Hosts derives at its own now, so a derivation between polls is
+// already right; only the wake waits for the poll.
+//
+// Off the refresh goroutine: the wake re-derives through Hosts() (its own
+// locks) and may touch the registry, and refreshKeys should not wait on it.
+func (m *homeMap) wakeOnMove(prev *homeSnapshot, now time.Time) {
+ snap := m.snapshot()
+ live := livePlatformHolds(snap, now)
+ m.mu.Lock()
+ moved := !samePlatformSet(prev.hosts, snap.hosts) || !sameNames(m.liveHolds, live)
+ m.liveHolds = live
wake := m.hostsWake
m.mu.Unlock()
- if wake != nil && (!samePlatformSet(prev.hosts, snap.hosts) || !sameNames(prev.platformHolds, snap.platformHolds)) {
- // The deployment's platform name set moved: a wildcard parent may
- // have appeared for an account homed elsewhere (#300). The set is
- // the ready fold's platform names plus the platform real holds, so
- // a platform hold arriving while the fold is unchanged wakes it
- // too. Off this goroutine — the wake re-derives through Hosts()
- // (its own locks) and may touch the registry, and refreshKeys
- // should not wait on that.
+ if wake != nil && moved {
go wake()
}
- return nil
+}
+
+// livePlatformHolds is the platform hold names whose exact resolution at now
+// is a platform holding page: the wildcard parent inputs (Hosts) the ready
+// fold does not carry. An expired hold is not one.
+func livePlatformHolds(snap *homeSnapshot, now time.Time) map[string]bool {
+ out := map[string]bool{}
+ for h := range snap.platformHolds {
+ if d := rank(h, snap.cands[h], now); d.kind == decHolding && d.platform {
+ out[h] = true
+ }
+ }
+ return out
}
// platformHolds is the names a remote real holding candidate marks
-// platform: wildcard parent inputs (Hosts) the ready fold does not carry.
+// platform: the candidates for livePlatformHolds.
func platformHolds(cands map[string][]candidate) map[string]bool {
out := map[string]bool{}
for name, cs := range cands {
@@ -589,12 +632,8 @@ func (m *homeMap) Hosts() []string {
seen[h] = true
out = append(out, h)
}
- now := m.clock()
- for h := range snap.platformHolds {
- if seen[h] {
- continue
- }
- if d := rank(h, snap.cands[h], now); d.kind == decHolding && d.platform {
+ for h := range livePlatformHolds(snap, m.clock()) {
+ if !seen[h] {
out = append(out, h)
}
}
diff --git a/internal/edge/homemap_test.go b/internal/edge/homemap_test.go
index 06938ba56..de5298eec 100644
--- a/internal/edge/homemap_test.go
+++ b/internal/edge/homemap_test.go
@@ -12,8 +12,10 @@ import (
"net/http"
"net/http/httptest"
"os"
+ "slices"
"strings"
"sync"
+ "sync/atomic"
"testing"
"time"
@@ -1991,3 +1993,162 @@ func TestFoldLegacyPublicationsUnchanged(t *testing.T) {
}
}
}
+
+// ---- freshness (self-serve moves plan 3, Task 5) ---------------------------
+//
+// A remote holding candidate is believed only while now.Sub(fetchedAt) <
+// holdingValidity; at exactly five minutes it is expired. fetchedAt is set by
+// refreshHosts on a successful parse of that file, and a carried-forward file
+// keeps the time it was last fetched at. Expiry is applied at ranking time
+// from the retained candidates, never by store I/O, and what answers after it
+// is whatever the remaining candidates give: the retained route, or the
+// wildcard (a local "*.<parent>" upstream or a remote one).
+
+// faultStore fails, on demand, the List, every Get, or the Get of named keys:
+// the ways one poll goes wrong. reads counts every List and Get, so a test can
+// show an answer changed without the store being asked anything.
+type faultStore struct {
+ release.Store
+ mu sync.Mutex
+ list bool
+ gets bool
+ keys map[string]bool
+ reads atomic.Int64
+}
+
+func (f *faultStore) set(list, gets bool, keys ...string) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.list, f.gets, f.keys = list, gets, map[string]bool{}
+ for _, k := range keys {
+ f.keys[k] = true
+ }
+}
+
+func (f *faultStore) List(ctx context.Context, prefix string) ([]string, error) {
+ f.reads.Add(1)
+ f.mu.Lock()
+ fail := f.list
+ f.mu.Unlock()
+ if fail {
+ return nil, errors.New("bucket down")
+ }
+ return f.Store.List(ctx, prefix)
+}
+
+func (f *faultStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ f.reads.Add(1)
+ f.mu.Lock()
+ fail := f.gets || f.keys[key]
+ f.mu.Unlock()
+ if fail {
+ return nil, "", errors.New("bucket down for this object")
+ }
+ return f.Store.Get(ctx, key)
+}
+
+// freshScene is a scene whose home map reads through a faultStore. put lands
+// a box's file in the bucket without polling; poll is one Refresh, its error
+// returned rather than fatal.
+type freshScene struct {
+ *scene
+ bucket release.Store
+ faults *faultStore
+}
+
+func newFreshScene(t *testing.T, self string) *freshScene {
+ t.Helper()
+ sc := newScene(t, self)
+ fs := &faultStore{Store: sc.m.s}
+ sc.m.s = fs
+ return &freshScene{scene: sc, bucket: fs.Store, faults: fs}
+}
+
+func edgesKey(box string) string { return edgesPrefix + box + ".json" }
+
+func (fx *freshScene) put(box, eip, hosts, offline string) {
+ fx.t.Helper()
+ putJSON(fx.t, fx.bucket, edgesKey(box), fmt.Sprintf(`{"v":1,"box":%q,"eip":%q,"hosts":{%s},"offline_moves":[%s]}`, box, eip, hosts, offline))
+}
+
+func (fx *freshScene) poll() error { return fx.m.Refresh(context.Background()) }
+
+func (fx *freshScene) mustPoll() {
+ fx.t.Helper()
+ if err := fx.poll(); err != nil {
+ fx.t.Fatalf("poll: %v", err)
+ }
+}
+
+// fetchedAt is the fetch time the map holds for box's file.
+func (fx *freshScene) fetchedAt(box string) time.Time {
+ return fx.m.snapshot().files[edgesKey(box)].fetchedAt
+}
+
+// at moves the clock to d after rankNow, the first poll's time.
+func (fx *freshScene) at(d time.Duration) { fx.now = rankNow.Add(d) }
+
+// Task 2 review Minor 2: the wildcard manager's parent set follows a remote
+// platform hold's clock. The wake compares the LIVE platform holds at the
+// poll's time, so the expiry itself wakes the manager (the parent leaves on
+// that derivation) and a later fresh fetch wakes it again (the parent
+// returns), though the files never changed and the ready fold never moved.
+func TestExpiredPlatformHoldLeavesTheParentSetAndRevives(t *testing.T) {
+ const apps = "apps.example.com"
+ held := "cart.bdj." + apps
+ fx := newFreshScene(t, "carol")
+ // The wake runs on its own goroutine and reads the clock: an atomic one.
+ var clock atomic.Int64
+ set := func(d time.Duration) { clock.Store(rankNow.Add(d).UnixNano()) }
+ set(0)
+ fx.m.now = func() time.Time { return time.Unix(0, clock.Load()).UTC() }
+ w := newWildcardManager(nil, apps, fx.e.deploymentHosts, nil, "", nil, nil)
+ var woke atomic.Int32
+ fx.m.setHostsWake(func() { w.refresh(); woke.Add(1) })
+ quiet := func(want int32, what string) {
+ t.Helper()
+ time.Sleep(20 * time.Millisecond)
+ if n := woke.Load(); n != want {
+ t.Fatalf("%s: woke %d time(s), want %d", what, n, want)
+ }
+ }
+
+ fx.put("annie", "1.1.1.1", readyJSON("shop.example.net", "bab", "shop", 0, 0), holdJSON(held, "mv_a", true, 1, 1))
+ fx.mustPoll()
+ waitWoke(t, &woke, 1)
+ waitParents(t, w, []string{"bdj." + apps})
+ readyFold := fx.m.snapshot().hosts
+
+ set(holdingValidity - time.Second)
+ fx.faults.set(true, false)
+ if fx.poll() == nil {
+ t.Fatal("test bug: the failed List must fail the poll")
+ }
+ quiet(1, "a live hold through a failed poll")
+
+ set(holdingValidity)
+ if fx.poll() == nil {
+ t.Fatal("test bug: the failed List must fail the poll")
+ }
+ waitWoke(t, &woke, 2)
+ waitParents(t, w, nil)
+ if slices.Contains(fx.m.Hosts(), held) {
+ t.Fatal("Hosts kept an expired hold")
+ }
+
+ set(holdingValidity + 30*time.Second)
+ fx.faults.set(false, true)
+ if fx.poll() == nil {
+ t.Fatal("test bug: every Get failing must fail the poll")
+ }
+ quiet(2, "an expired hold through another failed poll")
+
+ set(6 * time.Minute)
+ fx.faults.set(false, false)
+ fx.mustPoll()
+ if !samePlatformSet(readyFold, fx.m.snapshot().hosts) {
+ t.Fatal("test bug: the ready fold's platform set must not move")
+ }
+ waitWoke(t, &woke, 3)
+ waitParents(t, w, []string{"bdj." + apps})
+}