carlos / platform Public

zdd plan 1 task 7: cache entries tagged with serving route and generation

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev 5ee624339bf176523704a7d455a3ab5be51a4f84 parent 9b3c375
3 files changed, +264 −8
  • internal/edge/cache.go +49 −0
  • internal/edge/edge.go +28 −8
  • internal/edge/gencache_test.go +187 −0
diff --git a/internal/edge/cache.go b/internal/edge/cache.go
index d1c08cb16..0bebc2c0d 100644
--- a/internal/edge/cache.go
+++ b/internal/edge/cache.go
@@ -47,6 +47,15 @@ type CachedResponse struct {
Body []byte
StoredAt time.Time
TTL time.Duration
+ // Route and Gen tag an entry with the RESOLVED serving route (the
+ // instance host, whatever name the request used: an alias, a wildcard,
+ // a path-routed front door) and the generation of it that produced the
+ // response (zero-downtime spec §7.2). Empty Route: untagged, which is
+ // every entry with generations off and every route the activator does
+ // not manage. A tagged entry is live only while Gen is still the
+ // route's newest generation (SetGenerations).
+ Route string
+ Gen uint64
}
// Cache is an in-memory, per-host response cache for cacheable GET/HEAD
@@ -64,6 +73,10 @@ type Cache struct {
byURL map[cacheURLKey]map[cacheFullKey]*list.Element // urlKey -> its Vary variants
byPath map[cachePathKey]map[cacheFullKey]*list.Element // pathKey -> every entry under that path
byHost map[string]map[cacheFullKey]*list.Element // host -> every entry for that host
+ // genOf answers a tagged entry's route's newest generation
+ // (SetGenerations); nil with generations off. Called under mu, so it
+ // must take no lock its own side holds while calling into the cache.
+ genOf func(route string) (uint64, bool)
}
type cacheURLKey struct {
@@ -141,6 +154,30 @@ func NewCache(maxEntries, maxBodyBytes int) *Cache {
}
}
+// SetGenerations arms the generation guard (spec §7.2): a tagged entry is
+// refused at Put, and treated as a miss at lookup, unless its Gen is
+// genOf(Route). The edge calls it once, from EnableCache, only with
+// generations on.
+func (c *Cache) SetGenerations(genOf func(route string) (uint64, bool)) {
+ c.mu.Lock()
+ defer c.mu.Unlock()
+ c.genOf = genOf
+}
+
+// genCurrentLocked reports whether resp may be served or stored: untagged,
+// or tagged with its route's newest generation. A tagged entry with no
+// guard armed is not current: it cannot be vouched for. Caller holds c.mu.
+func (c *Cache) genCurrentLocked(resp *CachedResponse) bool {
+ if resp.Route == "" {
+ return true
+ }
+ if c.genOf == nil {
+ return false
+ }
+ n, ok := c.genOf(resp.Route)
+ return ok && n == resp.Gen
+}
+
func (k CacheKey) urlKey() cacheURLKey {
return cacheURLKey{Host: k.Host, Method: k.Method, Path: k.Path, RawQuery: k.RawQuery}
}
@@ -246,6 +283,12 @@ func (c *Cache) selectLocked(uk cacheURLKey, want func(*cacheEntry) bool) (*Cach
if !want(e) {
continue
}
+ // A generation that is no longer its route's newest is a miss for
+ // every name the route answers to (spec §7.2). Left in place: the
+ // switch's purge, the TTL or the LRU reclaims it.
+ if !c.genCurrentLocked(e.resp) {
+ continue
+ }
if bestEntry == nil || better(e, bestEntry) {
best, bestEntry = el, e
}
@@ -319,6 +362,12 @@ func (c *Cache) Put(key CacheKey, vary string, resp *CachedResponse) {
}
c.mu.Lock()
defer c.mu.Unlock()
+ // A response whose generation stopped being current while it was in
+ // flight (a late answer from a draining process) never fills the cache
+ // after the switch (spec §7.2).
+ if !c.genCurrentLocked(resp) {
+ return
+ }
fk := key.fullKey(canonicalVary(vary))
uk := fk.cacheURLKey
if el, ok := c.entries[fk]; ok {
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 27b7882cc..80772e355 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -405,6 +405,11 @@ func (e *Edge) EnableStatic(store release.Store, pub ed25519.PublicKey, cacheDir
// existing nil-means-disabled convention for Activator.
func (e *Edge) EnableCache(cache *Cache) {
e.cache = cache
+ // The generation guard (spec §7.2), only with generations on: e.gens is
+ // nil otherwise and the cache stays untagged, exactly as before.
+ if cache != nil && e.gens != nil {
+ cache.SetGenerations(e.gens.CacheGen)
+ }
}
// EnableTunnels wires the reverse-channel hub in: the edge answers hub.Host()
@@ -1743,7 +1748,7 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// The response is always recorded (even when not cacheable, in which
// case maybeStore is a no-op on rec.cacheable) so this branch stays a
// single shape regardless of method.
- rec := &recordingResponseWriter{ResponseWriter: w, request: r, cacheable: cacheable, maxBody: e.cacheMaxBody()}
+ rec := &recordingResponseWriter{ResponseWriter: w, request: r, cacheable: cacheable, genManaged: up.genManaged, maxBody: e.cacheMaxBody()}
if e.act.ServeHTTP(rec, r, up.route, up.proxy) {
e.maybeStore(key, rec)
return
@@ -1859,13 +1864,25 @@ func (e *Edge) maybeStore(key CacheKey, rec *recordingResponseWriter) {
// request — and so a response that varies by different headers, or doesn't
// vary at all, becomes a variant beside this one rather than a replacement
// for it.
- e.cache.Put(key, vary, &CachedResponse{
+ resp := &CachedResponse{
StatusCode: rec.status,
Header: rec.Header().Clone(),
Body: rec.body,
StoredAt: time.Now(),
TTL: ttl,
- })
+ }
+ // Tagged with the resolved serving route and its generation (spec
+ // §7.2). A managed response that no generation is recorded for (it
+ // reached the process before any generation was current) is not stored:
+ // nothing could later prove it current.
+ if rec.genManaged {
+ route, n, ok := servedGen(rec.request)
+ if !ok {
+ return
+ }
+ resp.Route, resp.Gen = route, n
+ }
+ e.cache.Put(key, vary, resp)
}
// hasTrailerSetCookie reports whether h carries a Set-Cookie TRAILER. An
@@ -2079,11 +2096,14 @@ type recordingResponseWriter struct {
http.ResponseWriter
request *http.Request
cacheable bool
- maxBody int
- status int
- body []byte
- truncated bool
- wroteHdr bool
+ // genManaged: the response came through appSocketProxy, so maybeStore
+ // stores it only tagged with the generation that served it.
+ genManaged bool
+ maxBody int
+ status int
+ body []byte
+ truncated bool
+ wroteHdr bool
}
func (r *recordingResponseWriter) WriteHeader(status int) {
diff --git a/internal/edge/gencache_test.go b/internal/edge/gencache_test.go
new file mode 100644
index 000000000..b71d03a4c
--- /dev/null
+++ b/internal/edge/gencache_test.go
@@ -0,0 +1,187 @@
+package edge
+
+import (
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// The guard at the cache's own level: a tagged entry is live only while its
+// generation is its route's newest, and a stale one is refused at Put.
+func TestCacheGenerationGuard(t *testing.T) {
+ c := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
+ cur := uint64(1)
+ c.SetGenerations(func(route string) (uint64, bool) { return cur, route == "inst.example.com" })
+ key := CacheKey{Host: "www.customer.example", Method: "GET", Path: "/"}
+ tagged := func(n uint64) *CachedResponse {
+ return &CachedResponse{StatusCode: 200, Body: []byte("x"), StoredAt: time.Now(), TTL: time.Minute, Route: "inst.example.com", Gen: n}
+ }
+ c.Put(key, "", tagged(1))
+ if _, ok := c.get(key); !ok {
+ t.Fatal("an entry tagged with the current generation missed")
+ }
+ cur = 2
+ if _, ok := c.get(key); ok {
+ t.Fatal("an entry from generation 1 hit after generation 2 became current")
+ }
+ other := CacheKey{Host: "inst.example.com", Method: "GET", Path: "/late"}
+ c.Put(other, "", tagged(1))
+ if len(c.byURL[other.urlKey()]) != 0 {
+ t.Fatal("Put stored a response from a generation that is no longer current")
+ }
+ c.Put(other, "", &CachedResponse{StatusCode: 200, Body: []byte("y"), StoredAt: time.Now(), TTL: time.Minute})
+ if _, ok := c.get(other); !ok {
+ t.Fatal("an untagged entry was refused; the guard must leave unmanaged routes alone")
+ }
+ unarmed := NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes)
+ unarmed.Put(key, "", tagged(2))
+ if _, ok := unarmed.get(key); ok {
+ t.Fatal("a tagged entry was served by a cache with no generation guard")
+ }
+}
+
+// genCacheFixture is genFixture's managed instance reached under four names
+// (its own host, a custom-domain alias, a wildcard alias, and a path-routed
+// front door), with the cache on and a second generation's app ready.
+func genCacheFixture(t *testing.T) (e *Edge, act *genActivator, a, b *genApp) {
+ t.Helper()
+ dir := genSockDir(t)
+ a = startGenApp(t, filepath.Join(dir, "a.sock"), body("A", "max-age=60"))
+ b = startGenApp(t, filepath.Join(dir, "b.sock"), body("B", "max-age=60"))
+ front := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { io.WriteString(w, "front") }))
+ t.Cleanup(front.Close)
+ reg := openReg(t)
+ for _, rt := range []registry.Route{
+ {Host: "inst.example.com", Kind: "instance", Hibernate: true, Socket: filepath.Join(dir, "canon.sock"), Status: "ready", Version: "v0"},
+ {Host: "front.example", Kind: "service", Addr: front.Listener.Addr().String(), Status: "ready"},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ upsertAlias(t, reg, "www.customer.example", "inst.example.com")
+ upsertAlias(t, reg, "*.wild.example", "inst.example.com")
+ upsertPath(t, reg, "front.example", "acct", "inst.example.com")
+ act = &genActivator{fakeGens: newFakeGens(true)}
+ act.switchTo("inst.example.com", GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ var err error
+ e, err = New(reg, act)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ e.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes))
+ return e, act, a, b
+}
+
+// Spec §12 "Cache aliases": a custom domain, a wildcard and a path-routed
+// request each miss after the switch, without anything listing the aliases.
+func TestCachedAliasesMissAfterTheSwitch(t *testing.T) {
+ e, act, _, b := genCacheFixture(t)
+ urls := []string{
+ "http://inst.example.com/p",
+ "http://www.customer.example/p",
+ "http://x.wild.example/p",
+ "http://front.example/acct/p",
+ }
+ for _, u := range urls {
+ if got := serve(t, e, u).Body.String(); got != "A" {
+ t.Fatalf("GET %s = %q, want A from generation 1", u, got)
+ }
+ }
+ before := act.served.Load()
+ for _, u := range urls {
+ if got := serve(t, e, u).Body.String(); got != "A" {
+ t.Fatalf("repeat GET %s = %q, want the cached A", u, got)
+ }
+ }
+ if act.served.Load() != before {
+ t.Fatal("repeat requests reached the activator; they should have been cache hits")
+ }
+ act.switchTo("inst.example.com", GenInfo{Socket: b.sock, N: 2, Build: "v2"})
+ for _, u := range urls {
+ rec := serve(t, e, u)
+ if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" {
+ t.Fatalf("GET %s after the switch = %q (version %q), want B from v2", u, rec.Body.String(), rec.Header().Get("X-Carlos-Version"))
+ }
+ }
+}
+
+// A response that finishes on the old generation after the switch is
+// returned to its client and never stored (spec §7.2 insertion rule).
+func TestLateResponseFromARetiredGenerationIsNotCached(t *testing.T) {
+ dir := genSockDir(t)
+ entered, gate := make(chan struct{}), make(chan struct{})
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ close(entered)
+ <-gate
+ w.Header().Set("Cache-Control", "max-age=60")
+ io.WriteString(w, "A")
+ }))
+ b := startGenApp(t, filepath.Join(dir, "b.sock"), body("B", "max-age=60"))
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{Host: "inst.example.com", Kind: "instance", Hibernate: true,
+ Socket: filepath.Join(dir, "canon.sock"), Status: "ready"}); err != nil {
+ t.Fatal(err)
+ }
+ act := &genActivator{fakeGens: newFakeGens(true)}
+ act.switchTo("inst.example.com", GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ e, err := New(reg, act)
+ if err != nil {
+ t.Fatal(err)
+ }
+ e.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes))
+
+ done := make(chan string)
+ go func() { done <- serve(t, e, "http://inst.example.com/slow").Body.String() }()
+ <-entered
+ act.switchTo("inst.example.com", GenInfo{Socket: b.sock, N: 2, Build: "v2"})
+ close(gate)
+ if got := <-done; got != "A" {
+ t.Fatalf("in-flight request got %q, want its own A", got)
+ }
+ if n := len(e.cache.byURL[cacheURLKey{Host: "inst.example.com", Method: "GET", Path: "/slow"}]); n != 0 {
+ t.Fatalf("%d entries stored for a response that finished after the switch, want 0", n)
+ }
+ if got := serve(t, e, "http://inst.example.com/slow").Body.String(); got != "B" {
+ t.Fatalf("next request = %q, want B", got)
+ }
+}
+
+// Cache-before-wake survives generations: a sleeping instance's cached page
+// still answers without a wake, because the entry's generation is still the
+// route's newest.
+func TestSleepingInstanceStillAnswersFromTheCache(t *testing.T) {
+ e, act, a, _ := genCacheFixture(t)
+ if got := serve(t, e, "http://inst.example.com/p").Body.String(); got != "A" {
+ t.Fatalf("first GET = %q, want A", got)
+ }
+ act.sleep("inst.example.com")
+ a.srv.Close()
+ before := act.served.Load()
+ if got := serve(t, e, "http://inst.example.com/p").Body.String(); got != "A" {
+ t.Fatalf("GET while asleep = %q, want the cached A", got)
+ }
+ if act.served.Load() != before {
+ t.Fatal("a cached page woke the instance")
+ }
+}
+
+// Flag off: entries are untagged and nothing about generations touches them.
+func TestGenerationsOffCachesUntagged(t *testing.T) {
+ e, act, _ := genFixture(t, false, true)
+ e.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes))
+ e.cache.Put(CacheKey{Host: "inst.example.com", Method: "GET", Path: "/p"}, "", &CachedResponse{
+ StatusCode: 200, Body: []byte("cached"), StoredAt: time.Now(), TTL: time.Minute})
+ act.switchTo("inst.example.com", GenInfo{Socket: "/nonexistent.sock", N: 9, Build: "v9"})
+ if got := serve(t, e, "http://inst.example.com/p").Body.String(); got != "cached" {
+ t.Fatalf("GET = %q, want the untagged cached body", got)
+ }
+ if e.cache.genOf != nil {
+ t.Fatal("the generation guard was armed with the switch off")
+ }
+}