amadan

carlos / platform Public

edge asset cache task 8 fix round 2: For answers from the last settled decision

For and Declared no longer read the registry per call. They answer from
an in-memory snapshot (host to HostAssets, plus every cached build's
declared paths) that the writer replaces only on a pass whose reads did
not overlap a records or alias pass, and that is computed lazily on
first use. While a pass runs the previous snapshot keeps answering with
its own at, so the home keeps serving cached assets without waking a
hibernating instance (design G4), and the one-hour rule still applies to
that at, so a box whose passes never settle stops catching.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 1a7552cb22b70bfa34106e21b2306f020ad0d49d parent cd20a0b
4 files changed, +296 −52
  • internal/host/assetmanifests.go +11 −0
  • internal/host/assetpolicy.go +100 −22
  • internal/host/assetpublish_test.go +180 −30
  • internal/host/edgesfile.go +5 −0
diff --git a/internal/host/assetmanifests.go b/internal/host/assetmanifests.go
index be81ddeee..4860a326f 100644
--- a/internal/host/assetmanifests.go
+++ b/internal/host/assetmanifests.go
@@ -79,6 +79,17 @@ func (c *assetManifestCache) prune(keep map[string]bool) {
}
}
+// all is a copy of every build held, by assetManifestKey.
+func (c *assetManifestCache) all() map[string][]string {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ out := make(map[string][]string, len(c.byKey))
+ for k, v := range c.byKey {
+ out[k] = slices.Clone(v)
+ }
+ return out
+}
+
// size is how many builds are held.
func (c *assetManifestCache) size() int {
c.mu.Lock()
diff --git a/internal/host/assetpolicy.go b/internal/host/assetpolicy.go
index 19651224e..afd28b3e0 100644
--- a/internal/host/assetpolicy.go
+++ b/internal/host/assetpolicy.go
@@ -8,6 +8,8 @@ import (
"fmt"
"slices"
"strings"
+ "sync"
+ "sync/atomic"
"time"
"github.com/carlosframework/platform/internal/edge"
@@ -112,6 +114,11 @@ type assetPolicy struct {
// betweenReads, when set, runs between a decision's path-rows read and
// its aliases read, the window a concurrent pass can tear. Tests only.
betweenReads func()
+
+ // snap is the last settled decision For and Declared answer from;
+ // lazyMu serializes computing the first one on first use.
+ snap atomic.Pointer[assetSnapshot]
+ lazyMu sync.Mutex
}
type writtenAssets struct {
@@ -296,48 +303,119 @@ func (p *assetPolicy) Publish(host string) (edgeAssets, bool) {
// For is the home's own catch for host (edge.LocalAssets): the rule without
// the pooled exclusion, answered only while the authority time is within
-// assetFreshFor. It reads the registry and never the bucket. A read that
-// overlapped a records or alias pass answers false: the request is served
-// uncached rather than from a mixture of the pass's before and after.
+// assetFreshFor.
+//
+// It answers from the last settled decision (assetSnapshot), never by
+// reading the registry or the bucket per call: while a records or alias pass
+// runs, the previous snapshot keeps answering with its own at, so the home
+// keeps serving cached assets and never wakes a hibernating instance for a
+// hit (design G4). The freshness rule applies to that at, so a box whose
+// passes never settle stops catching within the hour. Safe for concurrent
+// calls.
func (p *assetPolicy) For(host string) (edge.HostAssets, bool) {
- if p == nil {
+ s := p.current()
+ if s == nil {
return edge.HostAssets{}, false
}
- clocks := p.auth.snapshot()
- in, err := p.readAssetInputs()
- if err != nil || !p.auth.settled(clocks) {
- return edge.HostAssets{}, false
- }
- a, ok := p.decide(context.Background(), in, clocks, host, true, true)
+ a, ok := s.hosts[host]
if !ok || p.auth.now().Sub(a.At) > assetFreshFor {
return edge.HostAssets{}, false
}
+ a.Paths = slices.Clone(a.Paths)
return a, true
}
-// Declared reports whether build, of the app host is served by, declares a
-// prefix of path (edge.LocalAssets). It answers from the manifest cache
-// alone: a build not read yet declares nothing.
+// Declared reports whether build, of the app host catches for, declares a
+// prefix of path (edge.LocalAssets). It answers from the same snapshot as
+// For: a host the snapshot does not catch, or a build whose manifest had not
+// been read when the snapshot was taken, declares nothing.
func (p *assetPolicy) Declared(host, build, path string) bool {
- if p == nil || build == "" {
- return false
- }
- clocks := p.auth.snapshot()
- in, err := p.readAssetInputs()
- if err != nil || !p.auth.settled(clocks) {
+ if build == "" {
return false
}
- r, ok := in.primary(host)
- if !ok {
+ s := p.current()
+ if s == nil {
return false
}
- paths, ok := p.manifests.cached(release.RouteAccount(r), r.App, build)
+ a, ok := s.hosts[host]
if !ok {
return false
}
+ paths := s.builds[assetManifestKey(a.Account, a.App, build)]
return slices.ContainsFunc(paths, func(prefix string) bool { return strings.HasPrefix(path, prefix) })
}
+// assetSnapshot is one settled decision for every hostname this box could
+// catch for itself: For's answers (pooled routes included, aliases whatever
+// their DNS) and the declared paths of every build the manifest cache held.
+// It is immutable once published.
+type assetSnapshot struct {
+ hosts map[string]edge.HostAssets
+ builds map[string][]string
+}
+
+// current is the snapshot For and Declared answer from, computed on first
+// use when the writer has not published one yet. Nil when there is none and
+// none could be computed without overlapping a pass.
+func (p *assetPolicy) current() *assetSnapshot {
+ if p == nil {
+ return nil
+ }
+ if s := p.snap.Load(); s != nil {
+ return s
+ }
+ p.lazyMu.Lock()
+ defer p.lazyMu.Unlock()
+ if s := p.snap.Load(); s != nil {
+ return s
+ }
+ clocks := p.auth.snapshot()
+ in, err := p.readAssetInputs()
+ if err != nil || !p.auth.settled(clocks) {
+ return nil
+ }
+ s := p.buildSnapshot(context.Background(), in, clocks, true)
+ p.snap.CompareAndSwap(nil, s)
+ return p.snap.Load()
+}
+
+// buildSnapshot decides every route host and exact alias in in. cachedOnly
+// is the lazy path's (no bucket read off the writer's schedule). The caller
+// has checked that in was read settled against clocks.
+func (p *assetPolicy) buildSnapshot(ctx context.Context, in assetInputs, clocks assetClocks, cachedOnly bool) *assetSnapshot {
+ s := &assetSnapshot{hosts: map[string]edge.HostAssets{}}
+ consider := func(host string) {
+ if _, done := s.hosts[host]; done {
+ return
+ }
+ if a, ok := p.decide(ctx, in, clocks, host, true, cachedOnly); ok {
+ s.hosts[host] = a
+ }
+ }
+ for host := range in.byHost {
+ consider(host)
+ }
+ for alias := range in.aliasOf {
+ consider(alias)
+ }
+ s.builds = p.manifests.all()
+ return s
+}
+
+// publishSnapshot replaces the snapshot with one decided by the writer from
+// reads that did not overlap a pass. routes, paths and aliases are the
+// writer's own reads; the owners are recomputed over every served row,
+// because the home serves a host the claims withhold from the file.
+func (p *assetPolicy) publishSnapshot(ctx context.Context, routes []registry.Route, paths []registry.RoutePath, aliases []registry.RouteAlias, clocks assetClocks) {
+ served := make(map[string]edgeHost, len(routes))
+ for _, r := range routes {
+ if r.Status == "ready" && !r.IsSidecar() {
+ served[r.Host] = edgeHost{}
+ }
+ }
+ p.snap.Store(p.buildSnapshot(ctx, newAssetInputs(routes, served, paths, aliases, nil), clocks, false))
+}
+
// keepBuilds is the manifest cache's keep set: the installed and serving
// builds of every local route (design §4.1, R5-1).
func (p *assetPolicy) keepBuilds(routes []registry.Route) map[string]bool {
diff --git a/internal/host/assetpublish_test.go b/internal/host/assetpublish_test.go
index d06440caf..e98768f3f 100644
--- a/internal/host/assetpublish_test.go
+++ b/internal/host/assetpublish_test.go
@@ -1477,16 +1477,31 @@ func TestAssetDecisionOverlappingAnAliasPassWithholds(t *testing.T) {
t.Fatal("still due after a settled pass")
}
- // For and Publish: the same tear in the other direction, back to blog.
- // The old path set (shop's) with the new alias (to blog) would catch.
+ // For answers from the last settled decision and reads nothing, so the
+ // tear cannot reach it: the hook armed for the next read stays unfired.
+ // Publish reads, overlaps the hook's pass (back to blog) and refuses.
b.tearTo("blog.example.com")
+ fired := false
b.once(func() {
+ fired = true
if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
t.Error(err)
}
})
- if got, ok := b.policy.For("www.example.com"); ok {
- t.Fatalf("For answered %+v from a torn read", got)
+ if _, ok := b.policy.For("www.example.com"); ok {
+ t.Fatal("For caught the alias of a primary with path rows")
+ }
+ if _, ok := b.policy.For("blog.example.com"); !ok {
+ t.Fatal("For refused blog, which the last settled decision catches")
+ }
+ if fired {
+ t.Fatal("For read the registry")
+ }
+ if got, ok := b.policy.Publish("www.example.com"); ok {
+ t.Fatalf("Publish answered %+v from a torn read", got)
+ }
+ if !fired {
+ t.Fatal("Publish did not reach the hook: the fixture tested nothing")
}
aliases, err := b.reg.ListAliases()
if err != nil {
@@ -1495,23 +1510,18 @@ func TestAssetDecisionOverlappingAnAliasPassWithholds(t *testing.T) {
if !slices.ContainsFunc(aliases, func(a registry.RouteAlias) bool { return a.Alias == "www.example.com" && a.Host == "blog.example.com" }) {
t.Fatalf("the hook's pass did not move the alias: %+v", aliases)
}
- b.tearTo("shop.example.com")
- b.once(func() {
- if err := aliasPassReporting(ctx, b.doms, b.reg, nil, b.auth); err != nil {
- t.Error(err)
- }
- })
- if got, ok := b.policy.Publish("www.example.com"); ok {
- t.Fatalf("Publish answered %+v from a torn read", got)
- }
- // Settled again, the controls: blog has no path rows now, so blog
- // catches, and the alias (on shop, with path rows) does not.
+ // The next settled writer pass replaces the decision: blog now has path
+ // rows and shop has none.
b.policy.betweenReads = nil
- if _, ok := b.policy.For("blog.example.com"); !ok {
- t.Fatal("control: For refused blog on a settled read")
+ b.write()
+ if _, ok := b.policy.For("blog.example.com"); ok {
+ t.Fatal("For still caught blog after the settled pass gave it path rows")
+ }
+ if _, ok := b.policy.For("shop.example.com"); !ok {
+ t.Fatal("control: For refused shop on a settled decision")
}
if _, ok := b.policy.For("www.example.com"); ok {
- t.Fatal("For answered for the alias of a primary with path rows")
+ t.Fatal("For caught the alias of a primary with path rows")
}
}
@@ -1537,13 +1547,9 @@ func TestAssetDecisionOverlappingARecordsPassWithholds(t *testing.T) {
if got, ok := b.published("shop.example.com"); !ok || got.At != before.At {
t.Fatalf("an overlapped pass published %+v %v, want the last block at %d", got, ok, before.At)
}
- b.once(func() {
- if err := reconcileSerializedAuth(ctx, b.recs, b.reg, b.reconcileConfig(), nil, b.auth); err != nil {
- t.Error(err)
- }
- })
- if _, ok := b.policy.For("shop.example.com"); ok {
- t.Fatal("For answered across a records pass")
+ // For keeps answering from the last settled decision, with its at.
+ if got, ok := b.policy.For("shop.example.com"); !ok || got.At.Unix() != before.At {
+ t.Fatalf("For during the overlap = %+v %v, want the last settled block at %d", got, ok, before.At)
}
b.policy.betweenReads = nil
b.write()
@@ -1572,15 +1578,17 @@ func TestAssetDecisionWithholdsWhileAPassRuns(t *testing.T) {
w.confirm()
w.auth.passBegin()
wantAssets(t, w.hosts(), "shop.example.com", []string{"/assets/"}, assetGen1, assetT0)
- if _, ok := w.policy.For("shop.example.com"); ok {
- t.Fatal("For answered while a pass ran")
+ // The home's own catch keeps the last settled decision, with its at
+ // (design G4: a hit must not wake the instance because a pass runs).
+ if got, ok := w.policy.For("shop.example.com"); !ok || !got.At.Equal(assetT0) {
+ t.Fatalf("For while a pass ran = %+v %v, want the last settled block at %v", got, ok, assetT0)
+ }
+ if !w.policy.Declared("shop.example.com", "v1", "/assets/a.js") {
+ t.Fatal("Declared refused while a pass ran")
}
if _, ok := w.policy.Publish("shop.example.com"); ok {
t.Fatal("Publish answered while a pass ran")
}
- if w.policy.Declared("shop.example.com", "v1", "/assets/a.js") {
- t.Fatal("Declared answered while a pass ran")
- }
if !w.policy.rewriteDue() {
t.Fatal("no retry asked for")
}
@@ -1600,5 +1608,147 @@ func TestAssetDecisionWithholdsWhileAPassRuns(t *testing.T) {
wantNoAssets(t, w2.hosts(), "shop.example.com")
}
+// TestForAgesOutWhenPassesNeverSettle: a box stuck mid-pass keeps its last
+// settled decision, and For stops catching once that decision's at is over
+// an hour old.
+//
+// Not acceptance: the clocks are fed reports directly; the writer and the
+// reader are the production ones.
+func TestForAgesOutWhenPassesNeverSettle(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ w.pass()
+ w.auth.passBegin() // never ends
+ for range 12 {
+ w.clock.advance(5 * time.Minute)
+ w.confirm()
+ w.pass()
+ if _, ok := w.policy.For("shop.example.com"); !ok {
+ t.Fatalf("For refused at %v, inside the hour", w.clock.now())
+ }
+ }
+ w.clock.advance(time.Minute)
+ w.pass()
+ if got, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatalf("For answered %+v with a decision over an hour old", got)
+ }
+ if w.policy.Declared("shop.example.com", "v1", "/assets/a.js") != true {
+ t.Fatal("Declared is For's companion, not its gate: the build's declaration still reads")
+ }
+}
+
+// TestForMakesNoRegistryReadPerCall: once a decision exists, For and
+// Declared read nothing; the first call on a box whose writer has not run
+// yet computes one, and one that overlaps a pass is not kept.
+//
+// Not acceptance: the clocks are fed reports directly; the writer and the
+// reader are the production ones.
+func TestForMakesNoRegistryReadPerCall(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ reads := 0
+ w.policy.betweenReads = func() { reads++ }
+
+ // Lazy first use, overlapping a pass: nothing kept, nothing answered.
+ w.auth.passBegin()
+ if _, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatal("For answered from a decision that overlapped a pass")
+ }
+ if reads != 1 || w.policy.snap.Load() != nil {
+ t.Fatalf("reads = %d, snapshot kept = %v; want one read and none kept", reads, w.policy.snap.Load() != nil)
+ }
+ w.auth.passEnd()
+ // Lazy first use, settled: one read, then none. The manifest was never
+ // read (only the writer reads the bucket), so nothing catches yet.
+ if _, ok := w.policy.For("shop.example.com"); ok {
+ t.Fatal("the lazy decision caught a build whose manifest was never read")
+ }
+ if reads != 2 {
+ t.Fatalf("reads = %d after the lazy decision, want 2", reads)
+ }
+ w.pass() // the writer's settled pass replaces it (one more hooked read)
+ reads = 0
+ for range 100 {
+ if _, ok := w.policy.For("shop.example.com"); !ok {
+ t.Fatal("For refused after the writer's pass")
+ }
+ if !w.policy.Declared("shop.example.com", "v1", "/assets/x.js") {
+ t.Fatal("Declared refused after the writer's pass")
+ }
+ }
+ if reads != 0 {
+ t.Fatalf("200 calls made %d registry reads, want 0", reads)
+ }
+}
+
+// TestForIsSafeWhileTheWriterReplacesTheSnapshot runs For and Declared from
+// several goroutines while the writer decides, generations change and passes
+// overlap. Run with -race.
+//
+// Not acceptance: the clocks are fed reports directly; the writer and the
+// reader are the production ones.
+func TestForIsSafeWhileTheWriterReplacesTheSnapshot(t *testing.T) {
+ t.Parallel()
+ w := newAssetWorld(t)
+ w.ship("v1", "/assets/")
+ w.instance("shop.example.com", assetGen1, "v1")
+ w.confirm()
+ w.pass()
+ stop := make(chan struct{})
+ var wg sync.WaitGroup
+ for range 4 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ for {
+ select {
+ case <-stop:
+ return
+ default:
+ }
+ if a, ok := w.policy.For("shop.example.com"); ok && a.Gen != assetGen1 && a.Gen != assetGen2 {
+ t.Errorf("For answered an unknown generation %q", a.Gen)
+ }
+ w.policy.Declared("shop.example.com", "v1", "/assets/a.js")
+ }
+ }()
+ }
+ for i := range 40 {
+ gen := assetGen1
+ if i%2 == 1 {
+ gen = assetGen2
+ }
+ if err := w.reg.SetAssets("shop.example.com", false, gen); err != nil {
+ t.Fatal(err)
+ }
+ w.clock.advance(time.Minute)
+ w.confirm()
+ if i%3 == 0 {
+ w.auth.passBegin()
+ w.pass()
+ w.auth.passEnd()
+ } else {
+ w.pass()
+ }
+ }
+ close(stop)
+ wg.Wait()
+ // The last pass (i = 39, gen2) overlapped a pass, so the decision kept is
+ // i = 38's (gen1); the next settled pass moves it to gen2.
+ if got, ok := w.policy.For("shop.example.com"); !ok || got.Gen != assetGen1 {
+ t.Fatalf("after an overlapped last pass For = %+v %v, want the last settled generation %s", got, ok, assetGen1)
+ }
+ w.pass()
+ if got, ok := w.policy.For("shop.example.com"); !ok || got.Gen != assetGen2 {
+ t.Fatalf("after a settled pass For = %+v %v, want %s", got, ok, assetGen2)
+ }
+}
+
// Compile-time: the policy is what plan 3's edge consumes.
var _ edge.LocalAssets = (*assetPolicy)(nil)
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go
index 85c185371..25fd6a036 100644
--- a/internal/host/edgesfile.go
+++ b/internal/host/edgesfile.go
@@ -682,6 +682,11 @@ func edgesFileWriteClaims(ctx context.Context, s release.Store, reg *registry.Re
}
}
assets.manifests.prune(assets.keepBuilds(routes))
+ // The home's own catch (For, Declared) answers from the last
+ // settled decision; only a settled pass replaces it.
+ if settled {
+ assets.publishSnapshot(ctx, routes, paths, aliases, clocks)
+ }
}
var private BoxAddresses