amadan

carlos / platform Public

zdd plan 1 task 5: per-connection AppSocket dial and per-generation idle pools

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude a014ebdd92b2ea3788ae535f4464caef76ac8160 parent 43e325a
4 files changed, +793 −17
  • internal/edge/answered.go +21 −0
  • internal/edge/edge.go +1 −17
  • internal/edge/genproxy.go +353 −0
  • internal/edge/genproxy_test.go +418 −0
diff --git a/internal/edge/answered.go b/internal/edge/answered.go
index 2df6e81d9..9699a7a13 100644
--- a/internal/edge/answered.go
+++ b/internal/edge/answered.go
@@ -30,6 +30,11 @@ type outcome struct {
answered bool
cause string
start time.Time // when the edge began serving, so a proxy failure knows how long the client waited
+ // genHost and gen name the generation whose connection pool carried the
+ // request (genproxy.go), set only by an activator-managed route with
+ // generations on. The cache tags what it stores with them (spec §7.2).
+ genHost string
+ gen uint64
}
// causeClientGone is the cause noted when a proxy fails because the CLIENT
@@ -83,3 +88,19 @@ func noteCause(r *http.Request, cause string) {
o.cause = cause
}
}
+
+// noteServedGen records on r's outcome that generation n of host carried
+// the request; a no-op when r carries no outcome, like MarkAnswered.
+func noteServedGen(r *http.Request, host string, n uint64) {
+ if o, ok := r.Context().Value(answeredKey{}).(*outcome); ok {
+ o.genHost, o.gen = host, n
+ }
+}
+
+// servedGen is what noteServedGen recorded: ok false when nothing was.
+func servedGen(r *http.Request) (host string, n uint64, ok bool) {
+ if o, isOutcome := r.Context().Value(answeredKey{}).(*outcome); isOutcome && o.gen != 0 {
+ return o.genHost, o.gen, true
+ }
+ return "", 0, false
+}
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 4f95941a8..9f86a586a 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -59,7 +59,6 @@ import (
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
- "github.com/carlosframework/platform/internal/unixdial"
)
// Default cache sizing for EnableCache: 4096 entries is comfortably above a
@@ -2967,22 +2966,7 @@ func backhaulProxy(host, eip, box, auth string, tr http.RoundTripper, logThrottl
// in-flight request is never 502'd mid-deploy; a sustained-down backend hangs
// this long before 502 (rare: systemd Restart=always makes "down" transient).
func dialUnixWithRetry(ctx context.Context, d *net.Dialer, socketPath string) (net.Conn, error) {
- const window = 13 * time.Second
- deadline := time.Now().Add(window)
- for {
- conn, err := unixdial.Checked(ctx, d, socketPath)
- if err == nil {
- return conn, nil
- }
- if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) {
- return nil, err
- }
- select {
- case <-ctx.Done():
- return nil, ctx.Err()
- case <-time.After(100 * time.Millisecond):
- }
- }
+ return dialUnixRetrying(ctx, d, func() (string, bool) { return socketPath, true })
}
func isRetryableDialErr(err error) bool {
diff --git a/internal/edge/genproxy.go b/internal/edge/genproxy.go
new file mode 100644
index 000000000..bc43dcd53
--- /dev/null
+++ b/internal/edge/genproxy.go
@@ -0,0 +1,353 @@
+package edge
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httputil"
+ "sync"
+ "syscall"
+ "time"
+
+ "github.com/carlosframework/platform/internal/unixdial"
+)
+
+// genproxy.go is the edge half of generations (zero-downtime spec §7.2): a
+// reverse proxy for an activator-managed route whose app socket moves from
+// one generation to the next. Every request asks the activator for the
+// current generation and uses that generation's pool, whose connections all
+// reach that generation's socket (dialGen), so the response is attributed to
+// the generation that served it. A connection to generation n is never
+// reused for n+1, and a generation's idle connections are closed when it
+// stops being current.
+//
+// Used only when the activator's GenerationsOn is true. With it false the
+// edge builds unixProxy(t.Socket) exactly as before.
+
+// errNoGeneration is a dial's answer while the route has no current
+// generation (between a stop and the next wake). It wraps ENOENT on purpose:
+// to dialUnixRetrying it is the same "socket briefly absent" gap a restart
+// always had, retried inside the same window, and each retry asks again.
+var errNoGeneration = fmt.Errorf("edge: instance has no current generation: %w", syscall.ENOENT)
+
+// dialUnixRetrying is dialUnixWithRetry with the socket resolved per attempt:
+// resolve is asked before the first dial and again before every retry, so a
+// dial that races a switch lands on the generation that is current by then.
+func dialUnixRetrying(ctx context.Context, d *net.Dialer, resolve func() (string, bool)) (net.Conn, error) {
+ const window = 13 * time.Second
+ deadline := time.Now().Add(window)
+ for {
+ var (
+ conn net.Conn
+ err error
+ )
+ if sock, ok := resolve(); ok {
+ conn, err = unixdial.Checked(ctx, d, sock)
+ } else {
+ err = errNoGeneration
+ }
+ if err == nil {
+ return conn, nil
+ }
+ if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) {
+ return nil, err
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(100 * time.Millisecond):
+ }
+ }
+}
+
+// genMovedError is a generation pool's dial giving up: its own socket is
+// gone and another generation is current. Nothing was sent, so the request
+// can go again on the current generation's pool.
+type genMovedError struct{ to GenInfo }
+
+func (e *genMovedError) Error() string {
+ return fmt.Sprintf("edge: generation moved to %d while dialling the one before it", e.to.N)
+}
+
+// dialGen is a generation pool's dial: own's socket and nothing else, and
+// a connection proven to be own's (a shared foreign path is checked
+// against the current generation once connected), so every connection in
+// the pool is own's and a response through it is attributed to own exactly
+// (spec §7.2). The retiring generation still
+// accepts until its SIGTERM, which comes after the switch, so a request
+// that chose own's pool an instant before the switch is served there. A
+// retryable failure is retried inside the usual window while own is still
+// current or no generation is; once another one is current, own is gone for
+// good and the dial returns genMovedError.
+func dialGen(ctx context.Context, d *net.Dialer, src GenSource, host string, own GenInfo) (net.Conn, error) {
+ const window = 13 * time.Second
+ deadline := time.Now().Add(window)
+ for {
+ conn, err := unixdial.Checked(ctx, d, own.Socket)
+ if err == nil {
+ // Spec §7.2 (rev 17): if the generation stops being current
+ // before the dial completes, the request is retried against the
+ // new current generation. So every new connection is checked
+ // once connected: own must still be current. This also covers a
+ // foreign row, whose generations share one canonical path, so a
+ // connection on it could have reached the replacement. A
+ // connection already made (an idle one in own's pool) is not
+ // re-checked: the request that chose own's pool sampled own.
+ cur, ok := src.ServingGen(host)
+ if ok && cur.N == own.N {
+ return conn, nil
+ }
+ conn.Close()
+ if ok {
+ return nil, &genMovedError{to: cur}
+ }
+ err = errNoGeneration
+ }
+ if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) {
+ return nil, err
+ }
+ if cur, ok := src.ServingGen(host); ok && (cur.N != own.N || cur.Socket != own.Socket) {
+ return nil, &genMovedError{to: cur}
+ }
+ select {
+ case <-ctx.Done():
+ return nil, ctx.Err()
+ case <-time.After(100 * time.Millisecond):
+ }
+ }
+}
+
+// genPool is one generation's transport and the generation it reaches.
+type genPool struct {
+ tr *http.Transport
+ gen GenInfo
+}
+
+// GenPools is the per-socket-path set of transports, one idle pool per
+// generation socket. One set per consumer (the edge has one, scheduled
+// delivery has its own) and shared across route reloads, because the pool
+// belongs to the generation, not to the table build that first reached it.
+type GenPools struct {
+ mu sync.Mutex
+ m map[string]*genPool
+}
+
+// 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 {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ if gp, ok := p.m[g.Socket]; ok {
+ if gp.gen.N == g.N {
+ return gp
+ }
+ gp.tr.CloseIdleConnections()
+ }
+ gp := &genPool{tr: genTransportFor(host, src, &g), gen: g}
+ p.m[g.Socket] = gp
+ return gp
+}
+
+// Retire closes socket's idle connections and forgets its pool. It has the
+// shape of GenSource.OnGenRetired's hook. A request still in flight on a
+// retired pool finishes; its connection is not reused by anyone after.
+func (p *GenPools) Retire(_, socket string) {
+ p.mu.Lock()
+ gp, ok := p.m[socket]
+ delete(p.m, socket)
+ p.mu.Unlock()
+ if ok {
+ gp.tr.CloseIdleConnections()
+ }
+}
+
+// Len is the number of live pools.
+func (p *GenPools) Len() int {
+ p.mu.Lock()
+ defer p.mu.Unlock()
+ return len(p.m)
+}
+
+// genTransportFor builds the transport behind one pool: unixProxy's idle
+// bounds, and a dial that reaches own's socket only (dialGen). RoundTrip
+// always passes own: a request that finds no current generation waits for
+// one (waitServingGen) and then uses that generation's pool. With own nil
+// it is a one-shot transport whose dial resolves the socket per attempt.
+func genTransportFor(host string, src GenSource, own *GenInfo) *http.Transport {
+ dialer := &net.Dialer{}
+ dial := func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return dialUnixRetrying(ctx, dialer, func() (string, bool) { return src.AppSocket(host) })
+ }
+ if own != nil {
+ g := *own
+ dial = func(ctx context.Context, _, _ string) (net.Conn, error) {
+ return dialGen(ctx, dialer, src, host, g)
+ }
+ }
+ return &http.Transport{
+ DialContext: dial,
+ MaxIdleConns: 32,
+ IdleConnTimeout: 90 * time.Second,
+ DisableKeepAlives: own == nil,
+ }
+}
+
+// genRoundTripper picks the pool of host's current generation for each
+// request, stamps the response with the build of the generation whose
+// connection carried it, and records that generation on the request's
+// outcome for the cache (spec §7.2). A pool's connections are all its own
+// generation's (dialGen), so the pool's generation is that generation.
+type genRoundTripper struct {
+ host string
+ src GenSource
+ pools *GenPools
+}
+
+// genRetries bounds how often one request is sent again after its
+// generation moved under it.
+const genRetries = 2
+
+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
+ // behind heldBody: it reads it only once it is writing the request on a
+ // connection dialGen has proven to be the chosen generation's, and a
+ // Close it makes after a failed dial is held. So a dial that finds the
+ // generation moved leaves the body untouched, and the request goes again
+ // on the current generation's pool, a streaming POST with no GetBody
+ // included.
+ var hb *heldBody
+ if req.Body != nil && req.Body != http.NoBody {
+ hb = &heldBody{rc: req.Body}
+ req.Body = hb
+ }
+ for attempt := 0; ; attempt++ {
+ g, ok := t.src.ServingGen(t.host)
+ if !ok {
+ // No current generation (between a stop and the next wake): wait
+ // for one inside the dial's retry window, then take the normal
+ // path, so the response is attributed to the generation that
+ // serves it and its connection passes dialGen's check (spec §7.2).
+ g, ok = waitServingGen(req.Context(), t.src, t.host)
+ if !ok {
+ if hb != nil {
+ hb.release()
+ }
+ return nil, errNoGeneration
+ }
+ }
+ gp := t.pools.pool(t.host, g, t.src)
+ 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.
+ continue
+ }
+ if err != nil {
+ if hb != nil {
+ hb.release()
+ }
+ return nil, err
+ }
+ // X-Carlos-Version names the build of the generation that served the
+ // request, not the registry's Version (spec §7.2). Set, not Add: the
+ // platform's voice on this header is the only one.
+ if gp.gen.Build != "" {
+ resp.Header.Set("X-Carlos-Version", gp.gen.Build)
+ }
+ noteServedGen(req, t.host, gp.gen.N)
+ return resp, nil
+ }
+}
+
+// waitServingGen waits, within the dial's 13 s retry window and ctx, for
+// host to have a current generation, asking every 100 ms.
+func waitServingGen(ctx context.Context, src GenSource, host string) (GenInfo, bool) {
+ deadline := time.Now().Add(13 * time.Second)
+ for {
+ if g, ok := src.ServingGen(host); ok {
+ return g, true
+ }
+ if time.Now().After(deadline) {
+ return GenInfo{}, false
+ }
+ select {
+ case <-ctx.Done():
+ return GenInfo{}, false
+ case <-time.After(100 * time.Millisecond):
+ }
+ }
+}
+
+// heldBody is a request body the transport cannot consume or close before
+// the request is on a connection. The transport reads a body only while it
+// writes the request on a dialled connection, and closes it on any error,
+// including a dial that never connected. Until the first Read, that Close
+// is held, so the same body can go again on another generation's pool.
+// After it, Close passes through as usual.
+type heldBody struct {
+ rc io.ReadCloser
+ mu sync.Mutex
+ read bool
+}
+
+func (b *heldBody) Read(p []byte) (int, error) {
+ b.mu.Lock()
+ b.read = true
+ b.mu.Unlock()
+ return b.rc.Read(p)
+}
+
+func (b *heldBody) Close() error {
+ b.mu.Lock()
+ read := b.read
+ b.mu.Unlock()
+ if !read {
+ return nil // a failed dial's close: the body may still be sent
+ }
+ return b.rc.Close()
+}
+
+func (b *heldBody) wasRead() bool {
+ b.mu.Lock()
+ defer b.mu.Unlock()
+ return b.read
+}
+
+// release closes the body for good: the request has finished with an error.
+func (b *heldBody) release() { _ = b.rc.Close() }
+
+// appSocketProxy is unixProxy for an activator-managed route with
+// generations on: the same Rewrite, Host preservation, upgrade pass-through
+// and error handler, over genRoundTripper instead of one fixed socket.
+func appSocketProxy(host string, src GenSource, pools *GenPools) *httputil.ReverseProxy {
+ return &httputil.ReverseProxy{
+ Rewrite: func(pr *httputil.ProxyRequest) {
+ setForwarded(pr)
+ pr.Out.URL.Scheme = "http"
+ pr.Out.URL.Host = "carlos" // placeholder; the unix dialer ignores it
+ pr.Out.Host = pr.In.Host // backend sees the real hostname
+ scrubInboundPurgeHeader(pr)
+ },
+ Transport: &genRoundTripper{host: host, src: src, pools: pools},
+ ErrorHandler: defaultProxyError(proxyCause),
+ }
+}
+
+// AppSocketProxy is appSocketProxy for callers outside this package: the
+// scheduled-work deliverer in internal/host, which must reach the current
+// generation the same way the edge does (spec §7.1) without the edge's
+// cache, compression or usage layers.
+func AppSocketProxy(host string, src GenSource, pools *GenPools) http.Handler {
+ return appSocketProxy(host, src, pools)
+}
diff --git a/internal/edge/genproxy_test.go b/internal/edge/genproxy_test.go
new file mode 100644
index 000000000..594b55d08
--- /dev/null
+++ b/internal/edge/genproxy_test.go
@@ -0,0 +1,418 @@
+package edge
+
+import (
+ "context"
+ "errors"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "sync/atomic"
+ "syscall"
+ "testing"
+ "time"
+)
+
+// fakeGens is a GenSource a test drives by hand: switchTo is plan 3's
+// switch as the edge sees it, sleep is a stop.
+type fakeGens struct {
+ mu sync.Mutex
+ on bool
+ cur map[string]GenInfo // host -> current generation; absent = no current generation
+ last map[string]uint64 // host -> CacheGen
+ asks int // AppSocket calls
+ retired []func(host, socket string)
+ // onServing, when set, runs after every ServingGen answer, with no lock
+ // held: a test's way to land a switch between the pool choice and the
+ // dial.
+ onServing func()
+}
+
+func newFakeGens(on bool) *fakeGens {
+ return &fakeGens{on: on, cur: map[string]GenInfo{}, last: map[string]uint64{}}
+}
+
+func (f *fakeGens) GenerationsOn() bool { return f.on }
+
+func (f *fakeGens) AppSocket(host string) (string, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.asks++
+ g, ok := f.cur[host]
+ return g.Socket, ok
+}
+
+func (f *fakeGens) ServingGen(host string) (GenInfo, bool) {
+ f.mu.Lock()
+ g, ok := f.cur[host]
+ hook := f.onServing
+ f.mu.Unlock()
+ if hook != nil {
+ hook()
+ }
+ return g, ok
+}
+
+func (f *fakeGens) CacheGen(host string) (uint64, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ n, ok := f.last[host]
+ return n, ok
+}
+
+func (f *fakeGens) OnGenRetired(fn func(host, socket string)) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ f.retired = append(f.retired, fn)
+}
+
+// switchTo makes g host's current generation, then tells the retire hooks
+// about the one it replaced.
+func (f *fakeGens) switchTo(host string, g GenInfo) {
+ f.mu.Lock()
+ old, had := f.cur[host]
+ f.cur[host], f.last[host] = g, g.N
+ hooks := append([]func(string, string){}, f.retired...)
+ f.mu.Unlock()
+ if had {
+ for _, fn := range hooks {
+ fn(host, old.Socket)
+ }
+ }
+}
+
+// sleep takes host's current generation away and keeps its number for the
+// cache, as the activator does at a stop.
+func (f *fakeGens) sleep(host string) {
+ f.mu.Lock()
+ old, had := f.cur[host]
+ delete(f.cur, host)
+ hooks := append([]func(string, string){}, f.retired...)
+ f.mu.Unlock()
+ if had {
+ for _, fn := range hooks {
+ fn(host, old.Socket)
+ }
+ }
+}
+
+func (f *fakeGens) askCount() int {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return f.asks
+}
+
+// genSockDir is a short directory for generation sockets. Run with
+// TMPDIR=/tmp/k: a socket path must stay under sockaddr_un's 108 bytes.
+func genSockDir(t *testing.T) string {
+ t.Helper()
+ dir, err := os.MkdirTemp("", "e")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(dir) })
+ return dir
+}
+
+// genApp is one generation's process: an HTTP server on a unix socket that
+// counts the connections it opened and closed.
+type genApp struct {
+ sock string
+ ln net.Listener
+ opened, closed atomic.Int64
+ srv *http.Server
+}
+
+func startGenApp(t *testing.T, sock string, h http.Handler) *genApp {
+ t.Helper()
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ if errors.Is(err, syscall.EPERM) || strings.Contains(err.Error(), "operation not permitted") {
+ t.Skipf("unix-socket listeners not permitted in this environment: %v", err)
+ }
+ t.Fatalf("listen %s: %v", sock, err)
+ }
+ a := &genApp{sock: sock, ln: ln}
+ a.srv = &http.Server{Handler: h, ConnState: func(_ net.Conn, st http.ConnState) {
+ switch st {
+ case http.StateNew:
+ a.opened.Add(1)
+ case http.StateClosed, http.StateHijacked:
+ a.closed.Add(1)
+ }
+ }}
+ go a.srv.Serve(ln)
+ t.Cleanup(func() { a.srv.Close() })
+ return a
+}
+
+// stop is the generation's process exiting: the server and its listener close now,
+// so the socket file is unlinked before stop returns (srv.Close alone misses a
+// listener Serve has not registered yet, and Serve's late close would unlink the
+// path by name, possibly a replacement's socket).
+func (a *genApp) stop() { a.srv.Close(); a.ln.Close() }
+
+// body answers b with cache-control cc ("" for none).
+func body(b, cc string) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) {
+ if cc != "" {
+ w.Header().Set("Cache-Control", cc)
+ }
+ io.WriteString(w, b)
+ })
+}
+
+func serve(t *testing.T, h http.Handler, url string) *httptest.ResponseRecorder {
+ t.Helper()
+ rec := httptest.NewRecorder()
+ h.ServeHTTP(rec, httptest.NewRequest("GET", url, nil))
+ return rec
+}
+
+// Each retry asks the source again: a dial that races a switch reaches the
+// generation that is current by the time it retries.
+func TestDialUnixRetryingAsksAgainOnEveryRetry(t *testing.T) {
+ dir := genSockDir(t)
+ live := startGenApp(t, filepath.Join(dir, "2.sock"), body("B", ""))
+ var asks atomic.Int64
+ resolve := func() (string, bool) {
+ switch asks.Add(1) {
+ case 1:
+ return "", false // between generations: no current one yet
+ case 2:
+ return filepath.Join(dir, "1.sock"), true // a retired generation: its file is gone
+ default:
+ return live.sock, true
+ }
+ }
+ conn, err := dialUnixRetrying(context.Background(), &net.Dialer{}, resolve)
+ if err != nil {
+ t.Fatalf("dial: %v", err)
+ }
+ conn.Close()
+ if got := asks.Load(); got != 3 {
+ t.Fatalf("resolve asked %d times, want 3 (once per attempt)", got)
+ }
+}
+
+// A connection to generation n is never reused for n+1, and n's idle
+// connections close when it stops being current. The version header names
+// the build of the generation that served each request.
+func TestAppSocketProxyFollowsTheCurrentGeneration(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ b := startGenApp(t, filepath.Join(dir, "b.sock"), body("B", ""))
+ src := newFakeGens(true)
+ pools := NewGenPools()
+ src.OnGenRetired(pools.Retire)
+ const host = "inst.example.com"
+ src.switchTo(host, GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ p := appSocketProxy(host, src, pools)
+
+ for i := 0; i < 2; i++ {
+ rec := serve(t, p, "http://inst.example.com/")
+ if rec.Body.String() != "A" || rec.Header().Get("X-Carlos-Version") != "v1" {
+ t.Fatalf("request %d: %q version %q, want A from v1", i, rec.Body.String(), rec.Header().Get("X-Carlos-Version"))
+ }
+ }
+ if got := a.opened.Load(); got != 1 {
+ t.Fatalf("generation 1 saw %d connections, want 1 (keep-alive reused)", got)
+ }
+
+ src.switchTo(host, GenInfo{Socket: b.sock, N: 2, Build: "v2"})
+ rec := serve(t, p, "http://inst.example.com/")
+ if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" {
+ t.Fatalf("after the switch: %q version %q, want B from v2", rec.Body.String(), rec.Header().Get("X-Carlos-Version"))
+ }
+ if got := a.opened.Load(); got != 1 {
+ t.Fatalf("generation 1 saw %d connections after the switch, want still 1", got)
+ }
+ deadline := time.Now().Add(2 * time.Second)
+ for a.closed.Load() != 1 && time.Now().Before(deadline) {
+ time.Sleep(10 * time.Millisecond)
+ }
+ if got := a.closed.Load(); got != 1 {
+ t.Fatalf("generation 1's idle connection was not closed at retirement (closed=%d)", got)
+ }
+ if got := pools.Len(); got != 1 {
+ t.Fatalf("%d pools live, want 1 (the retired one forgotten)", got)
+ }
+}
+
+// Spec §7.2: the version header and the cache's generation name the
+// generation whose connection carried the request. Here the switch lands
+// between the pool choice and the dial. With a pool whose dial re-resolves
+// AppSocket, the request reached B while it was stamped and tagged as A.
+func TestAppSocketProxyAttributesTheConnectionsGeneration(t *testing.T) {
+ const host = "inst.example.com"
+ // Rev 17: a switch that lands before the dial completes sends the request
+ // to the new current generation, even though the retiring one still
+ // accepts. Its response is attributed to the generation that served it.
+ t.Run("a switch before the dial completes", func(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ b := startGenApp(t, filepath.Join(dir, "b.sock"), body("B", ""))
+ src := newFakeGens(true)
+ src.switchTo(host, GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ var once sync.Once
+ src.onServing = func() {
+ once.Do(func() { src.switchTo(host, GenInfo{Socket: b.sock, N: 2, Build: "v2"}) })
+ }
+ p := appSocketProxy(host, src, NewGenPools())
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ rec := httptest.NewRecorder()
+ p.ServeHTTP(rec, r)
+ _, n, _ := servedGen(r)
+ if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" || n != 2 {
+ t.Fatalf("served %q, version %q, generation %d: want B, v2, 2 (retried on the current generation)",
+ rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n)
+ }
+ if got := b.opened.Load(); got < 1 {
+ t.Fatal("generation 2 was never reached")
+ }
+ _ = a
+ })
+ // A proxied POST streams its body: no GetBody, unknown length. A switch
+ // before the dial completes must not turn it into a 502. The generation
+ // is settled before a byte of the body is read, so the body goes, whole,
+ // to the current generation. Retrying only replayable requests failed it.
+ t.Run("a streaming POST across a switch", func(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ echo := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ b, _ := io.ReadAll(r.Body)
+ io.WriteString(w, "B:"+string(b))
+ })
+ b := startGenApp(t, filepath.Join(dir, "b.sock"), echo)
+ src := newFakeGens(true)
+ src.switchTo(host, GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ var once sync.Once
+ src.onServing = func() {
+ once.Do(func() {
+ a.stop() // generation 1 has gone: its socket file with it
+ src.switchTo(host, GenInfo{Socket: b.sock, N: 2, Build: "v2"})
+ })
+ }
+ pr, pw := io.Pipe()
+ go func() {
+ io.WriteString(pw, "hello, ")
+ io.WriteString(pw, "world")
+ pw.Close()
+ }()
+ req := httptest.NewRequest("POST", "http://inst.example.com/form", pr)
+ req.ContentLength = -1
+ if req.GetBody != nil {
+ t.Fatal("fixture: the request must have no GetBody")
+ }
+ r, _ := withOutcome(req)
+ rec := httptest.NewRecorder()
+ appSocketProxy(host, src, NewGenPools()).ServeHTTP(rec, r)
+ if rec.Code != http.StatusOK || rec.Body.String() != "B:hello, world" || rec.Header().Get("X-Carlos-Version") != "v2" {
+ t.Fatalf("status %d, body %q, version %q: want 200, the whole body at generation 2, v2",
+ rec.Code, rec.Body.String(), rec.Header().Get("X-Carlos-Version"))
+ }
+ _ = b
+ })
+ t.Run("the sampled generation is gone", func(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ b := startGenApp(t, filepath.Join(dir, "b.sock"), body("B", ""))
+ src := newFakeGens(true)
+ src.switchTo(host, GenInfo{Socket: a.sock, N: 1, Build: "v1"})
+ var once sync.Once
+ src.onServing = func() {
+ once.Do(func() {
+ a.stop() // A has exited: its socket file is gone
+ src.switchTo(host, GenInfo{Socket: b.sock, N: 2, Build: "v2"})
+ })
+ }
+ p := appSocketProxy(host, src, NewGenPools())
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ rec := httptest.NewRecorder()
+ p.ServeHTTP(rec, r)
+ _, n, _ := servedGen(r)
+ if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" || n != 2 {
+ t.Fatalf("served %q, version %q, generation %d: want B, v2, 2 (sent again on the current pool)",
+ rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n)
+ }
+ })
+ // A foreign row's generations share one path. Generation 1's pool, chosen
+ // an instant before the hold, dials that path after generation 2 has
+ // bound it: the connection reaches 2, and a dial that trusts the path
+ // stamps 2's response as 1 and keeps the connection in 1's pool.
+ t.Run("a foreign row's path now belongs to the replacement", func(t *testing.T) {
+ dir := genSockDir(t)
+ path := filepath.Join(dir, "canonical.sock")
+ a := startGenApp(t, path, body("A", ""))
+ src := newFakeGens(true)
+ src.switchTo(host, GenInfo{Socket: path, N: 1, Build: "v1"})
+ var once sync.Once
+ var b *genApp
+ src.onServing = func() {
+ once.Do(func() {
+ a.stop() // the hold stops 1; its socket file goes
+ b = startGenApp(t, path, body("B", ""))
+ src.switchTo(host, GenInfo{Socket: path, N: 2, Build: "v2"})
+ })
+ }
+ p := appSocketProxy(host, src, NewGenPools())
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ rec := httptest.NewRecorder()
+ p.ServeHTTP(rec, r)
+ _, n, _ := servedGen(r)
+ if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" || n != 2 {
+ t.Fatalf("served %q, version %q, generation %d: want B, v2, 2 (never attributed to the old generation)",
+ rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n)
+ }
+ if b == nil || b.opened.Load() < 1 {
+ t.Fatal("the replacement was never reached")
+ }
+ })
+}
+
+// The generation that carried a request is recorded on its outcome, for the
+// cache to tag with.
+func TestAppSocketProxyNotesTheServedGeneration(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ src := newFakeGens(true)
+ src.switchTo("inst.example.com", GenInfo{Socket: a.sock, N: 7, Build: "v1"})
+ p := appSocketProxy("inst.example.com", src, NewGenPools())
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ p.ServeHTTP(httptest.NewRecorder(), r)
+ if host, n, ok := servedGen(r); !ok || host != "inst.example.com" || n != 7 {
+ t.Fatalf("servedGen = %q %d %v, want inst.example.com 7", host, n, ok)
+ }
+}
+
+// With no current generation the request waits inside the dial's window for
+// one, then goes the normal way: through that generation's pool, stamped
+// with its build and tagged with its number (spec §7.2). An unbound one-shot
+// transport returned the response unattributed, with no platform version.
+func TestAppSocketProxyWaitsForAGeneration(t *testing.T) {
+ dir := genSockDir(t)
+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))
+ src := newFakeGens(true)
+ pools := NewGenPools()
+ p := appSocketProxy("inst.example.com", src, pools)
+ go func() {
+ time.Sleep(300 * time.Millisecond)
+ src.switchTo("inst.example.com", GenInfo{Socket: a.sock, N: 3, Build: "v1"})
+ }()
+ r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil))
+ rec := httptest.NewRecorder()
+ p.ServeHTTP(rec, r)
+ if rec.Body.String() != "A" || rec.Header().Get("X-Carlos-Version") != "v1" {
+ t.Fatalf("body %q, version %q: want A from v1 once the generation appeared", rec.Body.String(), rec.Header().Get("X-Carlos-Version"))
+ }
+ if host, n, ok := servedGen(r); !ok || host != "inst.example.com" || n != 3 {
+ t.Fatalf("servedGen = %q %d %v, want generation 3", host, n, ok)
+ }
+ if pools.Len() != 1 {
+ t.Fatalf("%d pools, want generation 3's", pools.Len())
+ }
+}