amadan

carlos / platform Public

zdd plan 1 final review fix: a retired generation's pool is never recreated

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude eb9421595ea858fad99d3c42b6921f29105e7886 parent c8dbe3d
2 files changed, +135 −9
  • internal/edge/genproxy.go +45 −9
  • internal/edge/genproxy_test.go +90 −0
diff --git a/internal/edge/genproxy.go b/internal/edge/genproxy.go
index bc43dcd53..f876386b2 100644
--- a/internal/edge/genproxy.go
+++ b/internal/edge/genproxy.go
@@ -138,22 +138,38 @@ type GenPools struct {
// NewGenPools returns an empty set.
func NewGenPools() *GenPools { return &GenPools{m: map[string]*genPool{}} }
-// pool is generation g's pool, made on first use. Its dial reaches g's own
-// socket only (dialGen). A foreign row reuses its canonical path for every
-// generation (plan 1's reading), so a pool found for the path under another
-// generation number is replaced, and its idle connections closed.
-func (p *GenPools) pool(host string, g GenInfo, src GenSource) *genPool {
+// pool is generation g's pool, made on first use, or false when g is no
+// longer the host's current generation. Its dial reaches g's own socket only
+// (dialGen). A foreign row reuses its canonical path for every generation
+// (plan 1's reading), so a pool found for the path under another generation
+// number is replaced, and its idle connections closed.
+//
+// g was sampled by the caller some time ago. The check against src happens
+// under p.mu, and Retire takes the same lock, so a pool is published only for
+// a generation that was still current at that instant: if a switch has
+// retired g since, either Retire has already run (and this refuses to publish
+// a pool nothing would ever remove) or it runs after the publish and removes
+// it. The caller samples again on false.
+//
+// Lock order: p.mu, then whatever lock src.ServingGen takes (the activator's
+// own). That is safe because Retire runs from OnGenRetired hooks, which the
+// activator calls outside every lock of its own, and ServingGen never calls
+// into this set. A GenSource whose ServingGen reached Retire would deadlock.
+func (p *GenPools) pool(host string, g GenInfo, src GenSource) (*genPool, bool) {
p.mu.Lock()
defer p.mu.Unlock()
+ if cur, ok := src.ServingGen(host); !ok || cur.N != g.N {
+ return nil, false
+ }
if gp, ok := p.m[g.Socket]; ok {
if gp.gen.N == g.N {
- return gp
+ return gp, true
}
gp.tr.CloseIdleConnections()
}
gp := &genPool{tr: genTransportFor(host, src, &g), gen: g}
p.m[g.Socket] = gp
- return gp
+ return gp, true
}
// Retire closes socket's idle connections and forgets its pool. It has the
@@ -215,6 +231,12 @@ type genRoundTripper struct {
// generation moved under it.
const genRetries = 2
+// genStaleRetries bounds how often one request goes back to choosing a
+// generation because the one it chose was no longer current when its pool was
+// asked for. That is not a dial attempt, so it does not use genRetries, but a
+// source that keeps switching must not hold the request forever.
+const genStaleRetries = 3
+
func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
// The generation is chosen, and any switch retried, before a byte of the
// body is consumed (spec §7.2, rev 17). The transport gets the body
@@ -229,7 +251,7 @@ func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
hb = &heldBody{rc: req.Body}
req.Body = hb
}
- for attempt := 0; ; attempt++ {
+ for attempt, stale := 0, 0; ; {
g, ok := t.src.ServingGen(t.host)
if !ok {
// No current generation (between a stop and the next wake): wait
@@ -244,13 +266,27 @@ func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) {
return nil, errNoGeneration
}
}
- gp := t.pools.pool(t.host, g, t.src)
+ gp, ok := t.pools.pool(t.host, g, t.src)
+ if !ok {
+ // g was retired between the choice and the pool: choose again.
+ if stale++; stale > genStaleRetries {
+ if hb != nil {
+ hb.release()
+ }
+ if cur, ok := t.src.ServingGen(t.host); ok {
+ return nil, &genMovedError{to: cur}
+ }
+ return nil, errNoGeneration
+ }
+ continue
+ }
resp, err := gp.tr.RoundTrip(req)
var moved *genMovedError
if errors.As(err, &moved) && attempt < genRetries && (hb == nil || !hb.wasRead()) {
// The generation this request chose stopped being current before
// its dial completed. Nothing was sent and nothing of the body
// was read: send it again, on the current one.
+ attempt++
continue
}
if err != nil {
diff --git a/internal/edge/genproxy_test.go b/internal/edge/genproxy_test.go
index 594b55d08..35ec93dbf 100644
--- a/internal/edge/genproxy_test.go
+++ b/internal/edge/genproxy_test.go
@@ -9,6 +9,7 @@ import (
"net/http/httptest"
"os"
"path/filepath"
+ "strconv"
"strings"
"sync"
"sync/atomic"
@@ -416,3 +417,92 @@ func TestAppSocketProxyWaitsForAGeneration(t *testing.T) {
t.Fatalf("%d pools, want generation 3's", pools.Len())
}
}
+
+// A request that sampled generation n before a switch must not leave n's
+// pool behind when it resumes after Retire removed it: the pool is only
+// published for a generation that is still current. The switch lands at the
+// request's own sampling call, the instant between choosing n and asking for
+// its pool, and repeats over many generations; the set stays at most one pool
+// and every request is served by the generation current when it finished.
+func TestGenPoolsRetiredGenerationIsNotRecreated(t *testing.T) {
+ const (
+ host = "inst.example.com"
+ swaps = 20
+ )
+ dir := genSockDir(t)
+ apps := make([]*genApp, swaps+2)
+ for n := 1; n < len(apps); n++ {
+ apps[n] = startGenApp(t, filepath.Join(dir, strconv.Itoa(n)+".sock"), body("G"+strconv.Itoa(n), ""))
+ }
+ info := func(n int) GenInfo {
+ return GenInfo{Socket: apps[n].sock, N: uint64(n), Build: "v" + strconv.Itoa(n)}
+ }
+ src := newFakeGens(true)
+ pools := NewGenPools()
+ src.OnGenRetired(pools.Retire)
+ src.switchTo(host, info(1))
+ p := appSocketProxy(host, src, pools)
+
+ for n := 1; n <= swaps; n++ {
+ var once sync.Once
+ next := n + 1
+ src.onServing = func() { once.Do(func() { src.switchTo(host, info(next)) }) }
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ rec := httptest.NewRecorder()
+ p.ServeHTTP(rec, r)
+ want := "G" + strconv.Itoa(next)
+ if rec.Body.String() != want || rec.Header().Get("X-Carlos-Version") != "v"+strconv.Itoa(next) {
+ t.Fatalf("switch %d: served %q version %q, want %s from v%d",
+ n, rec.Body.String(), rec.Header().Get("X-Carlos-Version"), want, next)
+ }
+ if _, got, _ := servedGen(r); got != uint64(next) {
+ t.Fatalf("switch %d: attributed to generation %d, want %d", n, got, next)
+ }
+ if got := pools.Len(); got > 1 {
+ t.Fatalf("switch %d: %d pools live, want at most 1 (a retired generation's pool was recreated)", n, got)
+ }
+ }
+}
+
+// flappingGens answers a different generation on every ServingGen call, so no
+// sampled generation is ever still current when its pool is asked for.
+type flappingGens struct {
+ *fakeGens
+ sock string
+ n atomic.Uint64
+}
+
+func (f *flappingGens) ServingGen(string) (GenInfo, bool) {
+ return GenInfo{Socket: f.sock, N: f.n.Add(1), Build: "v"}, true
+}
+
+type closeCounter struct {
+ io.Reader
+ closed atomic.Int64
+}
+
+func (c *closeCounter) Close() error { c.closed.Add(1); return nil }
+
+// A request whose sampled generation is refused again and again does not spin
+// and does not leave a pool behind: after a few refusals it gives up with the
+// generation-moved error, as exhausted retries do, and its body is released.
+func TestGenRoundTripperGivesUpWhenEveryGenerationIsStale(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ src := &flappingGens{fakeGens: newFakeGens(true), sock: a.sock}
+ pools := NewGenPools()
+ rt := &genRoundTripper{host: "inst.example.com", src: src, pools: pools}
+ b := &closeCounter{Reader: strings.NewReader("x")}
+ req := httptest.NewRequest("POST", "http://inst.example.com/", b)
+ _, err := rt.RoundTrip(req)
+ var moved *genMovedError
+ if !errors.As(err, &moved) {
+ t.Fatalf("RoundTrip error = %v, want a *genMovedError", err)
+ }
+ if got := pools.Len(); got != 0 {
+ t.Fatalf("%d pools live, want none for a generation that was never current", got)
+ }
+ if got := b.closed.Load(); got != 1 {
+ t.Fatalf("body closed %d times, want 1", got)
+ }
+}