amadan

carlos / platform Public

edge: fleet routes ride the bound box's channel, fall over while dark

Task 4 of the customer-fleets plan: a route placed on a fleet
(Region == "fleet/<name>") now dispatches per the 1-2-3 rule — its bound
box's channel if connected, else any other connected channel of the
fleet (deterministic: TunnelHub.ConnectedOfFleet's sorted order), else a
clean 503 + Retry-After: 2 rather than the tunnel path's 502. Resolution
is per-request (transportFor pattern), so a reconnecting box needs no
Reload. Explicit rt.Tunnel keeps precedence; fleet routes share the
existing viaTunnel flag, so they're excluded from local hibernation
management exactly like -tunnel routes are today.
Paul Campbell d5878322843706922db46cc16eb26b312c1c4e6e parent 2ca9f44
3 files changed, +371 −1
  • internal/edge/edge.go +95 −1
  • internal/edge/fleet_route_test.go +255 −0
  • internal/edge/tunnel.go +21 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 53f2f00dd..599720865 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -237,6 +237,7 @@ 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
+ fleet, isFleetPlaced := registry.FleetPlacement(t.Region)
switch {
// Both reverse proxies get the Carlos-Purge response hook
// (purge_header.go): the app declares its own cache invalidations on
@@ -280,6 +281,21 @@ func (e *Edge) Reload() error {
p.ModifyResponse = e.purgeModifier(t)
proxy = p
viaTunnel = true
+ // Fleet placement (Region == "fleet/<name>", registry.FleetPlacement) is
+ // a customer-fleet route: its bytes leave the box over the SAME reverse
+ // channel a `-tunnel` binding uses, just resolved dynamically per
+ // request rather than to one fixed name — see fleetProxy. Guarded on
+ // e.tunnels for the identical reason the Tunnel case above is: a
+ // fleet-placed route on a hub-less edge degrades to its old Socket/Addr
+ // (or goes unserved) rather than the box silently going dark for a
+ // config gap that isn't its fault. viaTunnel is shared with the Tunnel
+ // case on purpose — a fleet-served instance is exactly as un-wakeable
+ // locally as a `-tunnel`-bound one; see the hibernation comment below.
+ case isFleetPlaced && e.tunnels != nil:
+ p := fleetProxy(fleet, t.FleetBox, e.tunnels)
+ p.ModifyResponse = e.purgeModifier(t)
+ proxy = p
+ viaTunnel = true
case t.Socket != "":
p := unixProxy(t.Socket)
p.ModifyResponse = e.purgeModifier(t)
@@ -308,7 +324,13 @@ func (e *Edge) Reload() error {
// local copy of the app beside the remote one. Leaving such a host out
// of `managed` is also what retires the local child a repointed route
// left running (RetireExcept, below) — the app moved to the remote
- // host, so the process it left behind here should go.
+ // host, so the process it left behind here should go. A fleet-placed
+ // route is the identical case (viaTunnel is set for both): whichever
+ // fleet box ends up serving it wakes its OWN local instance — that
+ // box's own activator, a separate process on a separate machine — so
+ // this platform edge must never manage it either. The lease race is
+ // the backstop if the fallback rule (fleetProxy) ever routes two
+ // requests to two different boxes for the same route in a gap.
hib := e.act != nil && t.Kind == "instance" && t.Hibernate && !viaTunnel
if hib {
managed[t.Host] = true
@@ -960,6 +982,78 @@ func tunnelProxy(name string, hub *TunnelHub) *httputil.ReverseProxy {
}
}
+// errFleetDark is the sentinel fleetProxy's Transport returns when fleet has
+// no connected channel at all — neither the route's bound box nor any
+// fallback. fleetProxy's ErrorHandler recognizes it and answers a clean 503
+// with Retry-After: 2 instead of the ReverseProxy default (a 502 with a
+// "context deadline exceeded"-flavored body that reads like a broken backend
+// rather than "no box is home right now, try again shortly").
+var errFleetDark = errors.New("edge: fleet has no connected channel")
+
+// fleetChannelFor picks which channel name to dial for one request against a
+// fleet-placed route, per the brief's binding/fallback rule:
+//
+// 1. boxLabel (the route's bound box, Route.FleetBox) if its channel
+// "<fleet>/<boxLabel>" is connected.
+// 2. else the first name in ConnectedOfFleet(fleet)'s sorted order — ANY
+// connected box of the fleet. That box's own activator wakes its
+// instance locally; a bound-but-different box racing it awake is the
+// lease's job to arbitrate, not this edge's.
+// 3. else "" — the fleet is dark.
+//
+// boxLabel may be "" (a route that wants a fleet but has no box assigned
+// yet), which just skips straight to the fallback step.
+func fleetChannelFor(fleet, boxLabel string, hub *TunnelHub) string {
+ if boxLabel != "" {
+ if bound := fleet + "/" + boxLabel; hub.State(bound) == "connected" {
+ return bound
+ }
+ }
+ if names := hub.ConnectedOfFleet(fleet); len(names) > 0 {
+ return names[0]
+ }
+ return ""
+}
+
+// fleetProxy reverse-proxies a fleet-placed route (Region == "fleet/<fleet>",
+// registry.FleetPlacement) over whichever channel fleetChannelFor picks —
+// resolved PER REQUEST, mirroring tunnelProxy/transportFor above, so a bound
+// box reconnecting (or a fallback box's channel dropping mid-outage) between
+// two requests is picked up with no Reload needed. Same Host preservation
+// and purge-header discipline as tunnelProxy; the only difference is the
+// no-channel-at-all case, which answers 503 + Retry-After via ErrorHandler
+// instead of falling into ReverseProxy's default 502 (see errFleetDark).
+func fleetProxy(fleet, boxLabel string, hub *TunnelHub) *httputil.ReverseProxy {
+ return &httputil.ReverseProxy{
+ Rewrite: func(pr *httputil.ProxyRequest) {
+ pr.SetXForwarded()
+ pr.Out.URL.Scheme = "https"
+ pr.Out.URL.Host = pr.In.Host // h2 :authority = the real hostname
+ pr.Out.Host = pr.In.Host
+ scrubInboundPurgeHeader(pr)
+ },
+ Transport: roundTripperFunc(func(req *http.Request) (*http.Response, error) {
+ name := fleetChannelFor(fleet, boxLabel, hub)
+ if name == "" {
+ return nil, errFleetDark
+ }
+ return hub.transportFor(name).RoundTrip(req)
+ }),
+ ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
+ if errors.Is(err, errFleetDark) {
+ w.Header().Set("Retry-After", "2")
+ http.Error(w, "fleet unreachable: no connected box", http.StatusServiceUnavailable)
+ return
+ }
+ // Any other transport failure (e.g. the picked channel died between
+ // fleetChannelFor's check and the RoundTrip call) — same shape as
+ // httputil.ReverseProxy's own default handler: log and 502.
+ log.Printf("edge: fleet %s proxy error: %v", fleet, err)
+ w.WriteHeader(http.StatusBadGateway)
+ },
+ }
+}
+
// dialUnixWithRetry masks the brief gap during an instance restart: while the
// old process has exited and the new one hasn't rebound the socket, dials fail
// with ENOENT/ECONNREFUSED. Retry those for a short window so no in-flight
diff --git a/internal/edge/fleet_route_test.go b/internal/edge/fleet_route_test.go
new file mode 100644
index 000000000..c27053362
--- /dev/null
+++ b/internal/edge/fleet_route_test.go
@@ -0,0 +1,255 @@
+package edge
+
+// Task 4: edge fleet-route dispatch — the 1-2-3 binding/fallback/dark rule
+// (task-4-brief.md):
+//
+// 1. Route.FleetBox != "" and channel "<fleet>/<FleetBox>" connected -> that
+// channel.
+// 2. else any connected channel of the fleet -> that one (deterministic:
+// ConnectedOfFleet's sorted order, first entry).
+// 3. else 503 with Retry-After: 2 (the fleet is dark).
+//
+// The bound/fallback cases need REAL h2 channels (a fake conn can't answer a
+// distinguishable response), so this uses the same tunnelclient.Loop-backed
+// fake remote host as tunnel_e2e_test.go rather than net.Pipe stand-ins.
+
+import (
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// namedHandler answers every request with name, so a test can tell which
+// fake remote host actually served a proxied request.
+func namedHandler(name string) http.Handler {
+ return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Write([]byte(name))
+ })
+}
+
+// fleetTestEdge builds an Edge + loopback server + armed hub over a registry
+// carrying fleet "acme" with the given boxes (label -> token) and the one
+// route, then wires fleet channel auth into the hub (Reload, same as
+// production's Watch loop). Returns the server URL and the hub so a test can
+// start fake remote hosts and wait on connection state.
+func fleetTestEdge(t *testing.T, boxes map[string]string, rt registry.Route) (string, *TunnelHub) {
+ t.Helper()
+ reg := openReg(t)
+ if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatalf("UpsertFleet: %v", err)
+ }
+ for label, token := range boxes {
+ if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: label, TokenHash: sha256Hex(token), Status: "active"}); err != nil {
+ t.Fatalf("AddFleetBox(%s): %v", label, err)
+ }
+ }
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert route: %v", err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ srv := httptest.NewServer(e)
+ t.Cleanup(srv.Close)
+ u, err := url.Parse(srv.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hub := NewTunnelHub(u.Hostname(), nil)
+ e.EnableTunnels(hub)
+ if err := e.Reload(); err != nil {
+ t.Fatalf("Reload (wire fleet auth): %v", err)
+ }
+ return srv.URL, hub
+}
+
+// getHostFull is getHost (edge_test.go) but returns the whole response, so a
+// test can also inspect headers (Retry-After).
+func getHostFull(t *testing.T, edgeURL, host string) *http.Response {
+ t.Helper()
+ req, err := http.NewRequest("GET", edgeURL+"/", nil)
+ if err != nil {
+ t.Fatalf("new request: %v", err)
+ }
+ req.Host = host
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("GET %s (host %s): %v", edgeURL, host, err)
+ }
+ return resp
+}
+
+// TestFleetRouteProxiesToBoundBox: rule 1 — a route bound (FleetBox) to a
+// box whose channel is connected rides that exact channel, even though a
+// second box of the same fleet is also connected and would sort first.
+func TestFleetRouteProxiesToBoundBox(t *testing.T) {
+ edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1", "pi-2": "tok2"}, registry.Route{
+ Host: "app.example", Kind: "instance", Status: "ready",
+ Region: registry.PlacementTarget("acme"), FleetBox: "pi-1",
+ })
+ stop1 := startFakeRemoteHost(t, edgeURL, "acme/pi-1", "tok1", namedHandler("pi-1"))
+ defer stop1()
+ stop2 := startFakeRemoteHost(t, edgeURL, "acme/pi-2", "tok2", namedHandler("pi-2"))
+ defer stop2()
+ waitForTunnelState(t, hub, "acme/pi-1", "connected", e2eReconnectBound)
+ waitForTunnelState(t, hub, "acme/pi-2", "connected", e2eReconnectBound)
+
+ code, body := getHost(t, edgeURL, "app.example")
+ if code != http.StatusOK || body != "pi-1" {
+ t.Fatalf("got %d %q, want 200 %q (the bound box, not the fallback)", code, body, "pi-1")
+ }
+}
+
+// TestFleetRouteFallsOverWhenBoundBoxIsDark: rule 2 — the bound box (pi-1)
+// never connects at all; the only connected channel of the fleet (pi-2)
+// serves the request instead of a 502/503.
+func TestFleetRouteFallsOverWhenBoundBoxIsDark(t *testing.T) {
+ edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1", "pi-2": "tok2"}, registry.Route{
+ Host: "app.example", Kind: "instance", Status: "ready",
+ Region: registry.PlacementTarget("acme"), FleetBox: "pi-1",
+ })
+ // pi-1 is never dialed — bound but dark.
+ stop2 := startFakeRemoteHost(t, edgeURL, "acme/pi-2", "tok2", namedHandler("pi-2"))
+ defer stop2()
+ waitForTunnelState(t, hub, "acme/pi-2", "connected", e2eReconnectBound)
+
+ if hub.State("acme/pi-1") == "connected" {
+ t.Fatal("test setup: pi-1 must not be connected for this case")
+ }
+ code, body := getHost(t, edgeURL, "app.example")
+ if code != http.StatusOK || body != "pi-2" {
+ t.Fatalf("got %d %q, want 200 %q (fallover to the only connected box)", code, body, "pi-2")
+ }
+}
+
+// TestFleetRouteDarkFleetAnswers503: rule 3 — no channel of the fleet is
+// connected at all (bound or otherwise): a clean 503 + Retry-After: 2, not
+// the tunnel path's 502.
+func TestFleetRouteDarkFleetAnswers503(t *testing.T) {
+ edgeURL, _ := fleetTestEdge(t, map[string]string{"pi-1": "tok1"}, registry.Route{
+ Host: "app.example", Kind: "instance", Status: "ready",
+ Region: registry.PlacementTarget("acme"), FleetBox: "pi-1",
+ })
+ // Nothing ever dials in.
+ resp := getHostFull(t, edgeURL, "app.example")
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("got %d, want 503", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Retry-After"); got != "2" {
+ t.Fatalf("Retry-After = %q, want %q", got, "2")
+ }
+}
+
+// TestFleetRouteWithNoBoundBoxUsesFallback: FleetBox can legitimately be ""
+// (placement wants a fleet but no box has been assigned yet) — that must
+// skip straight to the fallback rule rather than erroring on an empty bound
+// channel name.
+func TestFleetRouteWithNoBoundBoxUsesFallback(t *testing.T) {
+ edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1"}, registry.Route{
+ Host: "app.example", Kind: "instance", Status: "ready",
+ Region: registry.PlacementTarget("acme"), FleetBox: "",
+ })
+ stop1 := startFakeRemoteHost(t, edgeURL, "acme/pi-1", "tok1", namedHandler("pi-1"))
+ defer stop1()
+ waitForTunnelState(t, hub, "acme/pi-1", "connected", e2eReconnectBound)
+
+ code, body := getHost(t, edgeURL, "app.example")
+ if code != http.StatusOK || body != "pi-1" {
+ t.Fatalf("got %d %q, want 200 %q", code, body, "pi-1")
+ }
+}
+
+// TestExplicitTunnelPrecedesFleetPlacement: a route can carry both an
+// explicit Tunnel binding and a fleet Region ("carlos route -tunnel" doesn't
+// clear Region, mirroring the existing Tunnel-vs-Socket/Addr precedence) —
+// the explicit Tunnel wins, per the brief's "Explicit rt.Tunnel keeps
+// precedence (legacy/tito path unchanged)".
+func TestExplicitTunnelPrecedesFleetPlacement(t *testing.T) {
+ reg := openReg(t)
+ if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: sha256Hex("tok1"), Status: "active"}); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.Upsert(registry.Route{
+ Host: "app.example", Kind: "instance", Status: "ready",
+ Region: registry.PlacementTarget("acme"), FleetBox: "pi-1",
+ Tunnel: "legacy-box",
+ }); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+ u, err := url.Parse(srv.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hub := NewTunnelHub(u.Hostname(), map[string]string{"legacy-box": "sekrit"})
+ e.EnableTunnels(hub)
+ if err := e.Reload(); err != nil {
+ t.Fatal(err)
+ }
+ stop := startFakeRemoteHost(t, srv.URL, "legacy-box", "sekrit", namedHandler("legacy"))
+ defer stop()
+ waitForTunnelState(t, hub, "legacy-box", "connected", e2eReconnectBound)
+ // The fleet channel (acme/pi-1) is never dialed at all — if the fleet
+ // path were somehow taken, this would 503, not serve "legacy".
+ code, body := getHost(t, srv.URL, "app.example")
+ if code != http.StatusOK || body != "legacy" {
+ t.Fatalf("got %d %q, want 200 %q (explicit Tunnel must win over fleet placement)", code, body, "legacy")
+ }
+}
+
+// TestNonFleetRoutesUntouchedByFleetDispatch: an ordinary Addr-backed route
+// with no fleet placement at all (Region == "") must be completely unaffected
+// by the new dispatch case — same behavior as before Task 4.
+func TestNonFleetRoutesUntouchedByFleetDispatch(t *testing.T) {
+ backend, _ := hostEchoBackend(t, "plain backend")
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "plain.example", Kind: "service", Status: "ready",
+ Addr: backend.Listener.Addr().String(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+ code, body := getHost(t, srv.URL, "plain.example")
+ if code != http.StatusOK || body != "plain backend" {
+ t.Fatalf("got %d %q, want 200 %q", code, body, "plain backend")
+ }
+}
+
+// TestConnectedOfFleetSortedAndFiltered pins ConnectedOfFleet directly: only
+// names under "<fleet>/" come back, sorted, and a same-prefixed different
+// fleet ("acme2") is never mistaken for a member of "acme".
+func TestConnectedOfFleetSortedAndFiltered(t *testing.T) {
+ hub := NewTunnelHub("tunnel.test", nil)
+ for _, name := range []string{"acme/pi-2", "acme/pi-1", "acme2/pi-1", "other/pi-1"} {
+ c1, c2 := net.Pipe()
+ t.Cleanup(func() { c2.Close() })
+ hub.mu.Lock()
+ hub.conns[name] = &tunnelConn{raw: c1}
+ hub.mu.Unlock()
+ }
+ got := hub.ConnectedOfFleet("acme")
+ want := []string{"acme/pi-1", "acme/pi-2"}
+ if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] {
+ t.Fatalf("ConnectedOfFleet(acme) = %v, want %v", got, want)
+ }
+}
diff --git a/internal/edge/tunnel.go b/internal/edge/tunnel.go
index d73f4eb50..fc9eda944 100644
--- a/internal/edge/tunnel.go
+++ b/internal/edge/tunnel.go
@@ -17,6 +17,7 @@ import (
"log"
"net"
"net/http"
+ "slices"
"strings"
"sync"
"time"
@@ -302,6 +303,26 @@ func (h *TunnelHub) State(name string) string {
return ""
}
+// ConnectedOfFleet returns every currently-connected channel name belonging
+// to fleet — the "<fleet>/<label>" names whose prefix matches — sorted. This
+// is what makes the edge's fleet-route fallback (edge.go's fleetProxy)
+// deterministic: "any connected box of this fleet" always picks the same one
+// for a given live set, first entry, rather than whatever order Go's map
+// iteration happens to produce.
+func (h *TunnelHub) ConnectedOfFleet(fleet string) []string {
+ prefix := fleet + "/"
+ h.mu.Lock()
+ defer h.mu.Unlock()
+ var out []string
+ for name := range h.conns {
+ if strings.HasPrefix(name, prefix) {
+ out = append(out, name)
+ }
+ }
+ slices.Sort(out)
+ return out
+}
+
type roundTripperFunc func(*http.Request) (*http.Response, error)
func (f roundTripperFunc) RoundTrip(r *http.Request) (*http.Response, error) { return f(r) }