carlos / platform Public

zdd plan 1 task 6: edge proxies managed routes through the current generation

Co-Authored-By: Claude Haiku 4.5 <noreply@anthropic.com>

claude 2 weeks ago 9b3c3759d6f845a44d53462c302e79b74bee902b parent a014ebd

2 files changed, +136 −3

internal/edge/edge.go +38 −3
internal/edge/genedge_test.go +98 −0

diff --git a/internal/edge/edge.go b/internal/edge/edge.go

index 9f86a586a..27b7882cc 100644

--- a/internal/edge/edge.go

+++ b/internal/edge/edge.go

@@ -275,6 +275,14 @@ type Edge struct {

doorAddr string

doorKeys *doorKeyMirror

closeDoor func()

+ // gens is the activator's generation view (zero-downtime spec §7.2),

+ // set in New only when the activator implements GenSource AND its

+ // switch is on; genPools is the edge's per-generation-socket idle

+ // pools, retired through gens.OnGenRetired. Both nil with the switch off

+ // (CARLOS_SWAP unset), and then nothing below reads them: managed

+ // routes get unixProxy(t.Socket) and the row's version, as before.

+ gens GenSource

+ genPools *GenPools

}

// RunOption configures the Edge that Run/RunWithActivator* builds, before it

@@ -348,6 +356,10 @@ type upstream struct {

// captured here would be a stale copy of the target's entry the moment

// the target's route changed within the same map build.

paths map[string]string

+ // genManaged marks an activator-managed instance proxied through

+ // appSocketProxy (generations on). Its cacheable responses are stored

+ // only when tagged with the generation that served them (cache.go).

+ genManaged bool

}

// New builds an Edge over reg and performs the startup reload. act may be nil

@@ -356,6 +368,12 @@ func New(reg *registry.Registry, act Activator) (*Edge, error) {

e := &Edge{reg: reg, act: act, gate: gate.New(), probers: newProberSet(), gzipOn: gzipEnabled(), steeringHost: steeringProbeHostFromEnv(), appsDomain: appsDomainFromEnv()}

e.geo = geo.NewSwappable()

e.geoLookup = e.geo.Lookup

+ // Generations (zero-downtime spec §7.2), before the startup Reload so

+ // its first table build already proxies through them.

+ if gs, ok := act.(GenSource); ok && gs.GenerationsOn() {

+ e.gens, e.genPools = gs, NewGenPools()

+ gs.OnGenRetired(e.genPools.Retire)

+ }

if err := e.Reload(); err != nil {

return nil, err

}

@@ -621,6 +639,9 @@ func (e *Edge) Reload() error {

// viaTunnel: this host's bytes leave the box over a reverse channel,

// which is what disqualifies it from hibernation below.

viaTunnel := false

+ // genManaged: an activator-managed instance reached through its

+ // current generation (appSocketProxy), generations on only.

+ genManaged := false

fleet, isFleetPlaced := registry.FleetPlacement(t.Region)

switch {

// Both reverse proxies get the Carlos-Purge response hook

@@ -681,7 +702,17 @@ func (e *Edge) Reload() error {

proxy = p

viaTunnel = true

case t.Socket != "":

- p := unixProxy(t.Socket)

+ // An activator-managed instance's process binds a generation

+ // socket, not t.Socket, once generations are on (spec §7.1):

+ // every new connection asks the activator for the current one.

+ // t.Kind/t.Hibernate are the activator-managed test below

+ // (hib); viaTunnel is false in this arm by construction.

+ var p *httputil.ReverseProxy

+ if e.gens != nil && t.Kind == "instance" && t.Hibernate {

+ p, genManaged = appSocketProxy(t.Host, e.gens, e.genPools), true

+ } else {

+ p = unixProxy(t.Socket)

+ }

p.ModifyResponse = e.responseModifier(t)

proxy = p

// TCP upstreams. The gate is AddrList (registry), not t.Addr, so a

@@ -771,7 +802,11 @@ func (e *Edge) Reload() error {

// wrapping those would put a version on a response that is serving

// none, which is exactly the false "verified" the header exists to

// prevent.

- if t.Version != "" && t.Kind != "static" && t.Kind != "status" {

+ //

+ // A genManaged route is excluded too: its header comes from the

+ // generation that served the request (genRoundTripper, spec §7.2),

+ // and the registry's Version names only what is installed.

+ if t.Version != "" && t.Kind != "static" && t.Kind != "status" && !genManaged {

proxy = versionHeader(t.Version, proxy)

}

// Hibernation is per-host: only an instance whose route carries the

@@ -832,7 +867,7 @@ func (e *Edge) Reload() error {

"route %s is a hibernating instance but this box has no activator, so nothing here will create %s or start its process. If this box runs `carlos agent`, set CARLOS_HIBERNATE_BUCKET in /etc/carlos/host.env and restart carlos-edge; `carlos edge` on its own never runs instances.",

t.Host, t.Socket))

}

- m[t.Host] = upstream{proxy: proxy, route: t, hibernating: hib, viaTunnel: viaTunnel}

+ m[t.Host] = upstream{proxy: proxy, route: t, hibernating: hib, viaTunnel: viaTunnel, genManaged: genManaged}

}

// Path routes, folded BEFORE aliases so an alias of a prefix-routed

// hostname inherits its map along with everything else it copies — an

diff --git a/internal/edge/genedge_test.go b/internal/edge/genedge_test.go

new file mode 100644

index 000000000..4f28d8b70

--- /dev/null

+++ b/internal/edge/genedge_test.go

@@ -0,0 +1,98 @@

+package edge

+import (

+ "net/http"

+ "path/filepath"

+ "sync/atomic"

+ "testing"

+ "github.com/carlosframework/platform/internal/registry"

+// genActivator is an Activator that is also a GenSource: an instant wake

+// that proxies straight through, over a hand-driven fakeGens. served counts

+// the requests that reached it, so a test can tell a cache hit (none) from

+// a proxied request.

+type genActivator struct {

+ *fakeGens

+ served atomic.Int64

+func (a *genActivator) ServeHTTP(w http.ResponseWriter, r *http.Request, _ registry.Route, proxy http.Handler) bool {

+ a.served.Add(1)

+ proxy.ServeHTTP(w, r)

+ return true

+func (a *genActivator) RetireExcept(map[string]bool) {}

+// genFixture is a registry with one activator-managed instance whose

+// CANONICAL socket is served by an app answering "canonical", and a

+// generation-1 app answering "A". Which one a request reaches says which

+// path the edge took.

+func genFixture(t *testing.T, on bool, hibernate bool) (*Edge, *genActivator, *genApp) {

+ t.Helper()

+ dir := genSockDir(t)

+ startGenApp(t, filepath.Join(dir, "canon.sock"), body("canonical", ""))

+ a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", ""))

+ reg := openReg(t)

+ if err := reg.Upsert(registry.Route{Host: "inst.example.com", Kind: "instance", Hibernate: hibernate,

+ Socket: filepath.Join(dir, "canon.sock"), Status: "ready", Version: "v0"}); err != nil {

+ t.Fatal(err)

+ }

+ act := &genActivator{fakeGens: newFakeGens(on)}

+ act.switchTo("inst.example.com", GenInfo{Socket: a.sock, N: 1, Build: "v1"})

+ e, err := New(reg, act)

+ if err != nil {

+ t.Fatalf("New: %v", err)

+ }

+ return e, act, a

+func TestManagedRouteDialsTheCurrentGeneration(t *testing.T) {

+ e, act, _ := genFixture(t, true, true)

+ rec := serve(t, e, "http://inst.example.com/")

+ if rec.Body.String() != "A" {

+ t.Fatalf("body %q, want A: the edge dialled the canonical socket, not the current generation", rec.Body.String())

+ }

+ if got := rec.Header().Get("X-Carlos-Version"); got != "v1" {

+ t.Fatalf("X-Carlos-Version = %q, want v1, the serving generation's build, not the row's v0", got)

+ }

+ if !e.current()["inst.example.com"].genManaged {

+ t.Fatal("the managed route's upstream is not marked genManaged")

+ }

+ if len(act.retired) != 1 {

+ t.Fatalf("New registered %d retire hooks, want 1 (the edge's pools)", len(act.retired))

+ }

+// Flag off: byte for byte the old edge. unixProxy on the canonical socket,

+// the row's version, no retire hook, no pools.

+func TestGenerationsOffKeepsTheCanonicalProxyAndTheRowVersion(t *testing.T) {

+ e, act, _ := genFixture(t, false, true)

+ rec := serve(t, e, "http://inst.example.com/")

+ if rec.Body.String() != "canonical" || rec.Header().Get("X-Carlos-Version") != "v0" {

+ t.Fatalf("got %q version %q, want canonical v0", rec.Body.String(), rec.Header().Get("X-Carlos-Version"))

+ }

+ if e.gens != nil || e.genPools != nil || len(act.retired) != 0 {

+ t.Fatal("generations were wired with the switch off")

+ }

+ up := e.current()["inst.example.com"]

+ if up.genManaged {

+ t.Fatal("upstream marked genManaged with the switch off")

+ }

+// An unflagged socket route is an externally supervised process: canonical,

+// whatever the switch says.

+func TestUnmanagedSocketRouteStaysCanonicalWithGenerationsOn(t *testing.T) {

+ e, _, _ := genFixture(t, true, false)

+ rec := serve(t, e, "http://inst.example.com/")

+ if rec.Body.String() != "canonical" || rec.Header().Get("X-Carlos-Version") != "v0" {

+ t.Fatalf("got %q version %q, want canonical v0", rec.Body.String(), rec.Header().Get("X-Carlos-Version"))

+ }

+ up := e.current()["inst.example.com"]

+ if up.genManaged {

+ t.Fatal("an unflagged route was marked genManaged")

+ }