| 1 | package edge |
| 2 | |
| 3 | // Task 4: edge fleet-route dispatch — the 1-2-3 binding/fallback/dark rule |
| 4 | // (task-4-brief.md): |
| 5 | // |
| 6 | // 1. Route.FleetBox != "" and channel "<fleet>/<FleetBox>" connected -> that |
| 7 | // channel. |
| 8 | // 2. else any connected channel of the fleet -> that one (deterministic: |
| 9 | // ConnectedOfFleet's sorted order, first entry). |
| 10 | // 3. else 503 with Retry-After: 2 (the fleet is dark). |
| 11 | // |
| 12 | // The bound/fallback cases need REAL h2 channels (a fake conn can't answer a |
| 13 | // distinguishable response), so this uses the same tunnelclient.Loop-backed |
| 14 | // fake remote host as tunnel_e2e_test.go rather than net.Pipe stand-ins. |
| 15 | |
| 16 | import ( |
| 17 | "net" |
| 18 | "net/http" |
| 19 | "net/http/httptest" |
| 20 | "net/url" |
| 21 | "testing" |
| 22 | |
| 23 | "github.com/carlosframework/platform/internal/registry" |
| 24 | ) |
| 25 | |
| 26 | // namedHandler answers every request with name, so a test can tell which |
| 27 | // fake remote host actually served a proxied request. |
| 28 | func namedHandler(name string) http.Handler { |
| 29 | return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| 30 | w.Write([]byte(name)) |
| 31 | }) |
| 32 | } |
| 33 | |
| 34 | // fleetTestEdge builds an Edge + loopback server + armed hub over a registry |
| 35 | // carrying fleet "acme" with the given boxes (label -> token) and the one |
| 36 | // route, then wires fleet channel auth into the hub (Reload, same as |
| 37 | // production's Watch loop). Returns the server URL and the hub so a test can |
| 38 | // start fake remote hosts and wait on connection state. |
| 39 | func fleetTestEdge(t *testing.T, boxes map[string]string, rt registry.Route) (string, *TunnelHub) { |
| 40 | t.Helper() |
| 41 | reg := openReg(t) |
| 42 | if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil { |
| 43 | t.Fatalf("UpsertFleet: %v", err) |
| 44 | } |
| 45 | for label, token := range boxes { |
| 46 | if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: label, TokenHash: sha256Hex(token), Status: "active"}); err != nil { |
| 47 | t.Fatalf("AddFleetBox(%s): %v", label, err) |
| 48 | } |
| 49 | } |
| 50 | if err := reg.Upsert(rt); err != nil { |
| 51 | t.Fatalf("upsert route: %v", err) |
| 52 | } |
| 53 | e, err := New(reg, nil) |
| 54 | if err != nil { |
| 55 | t.Fatalf("New: %v", err) |
| 56 | } |
| 57 | srv := httptest.NewServer(e) |
| 58 | t.Cleanup(srv.Close) |
| 59 | u, err := url.Parse(srv.URL) |
| 60 | if err != nil { |
| 61 | t.Fatal(err) |
| 62 | } |
| 63 | hub := NewTunnelHub(u.Hostname(), nil) |
| 64 | e.EnableTunnels(hub) |
| 65 | if err := e.Reload(); err != nil { |
| 66 | t.Fatalf("Reload (wire fleet auth): %v", err) |
| 67 | } |
| 68 | return srv.URL, hub |
| 69 | } |
| 70 | |
| 71 | // getHostFull is getHost (edge_test.go) but returns the whole response, so a |
| 72 | // test can also inspect headers (Retry-After). |
| 73 | func getHostFull(t *testing.T, edgeURL, host string) *http.Response { |
| 74 | t.Helper() |
| 75 | req, err := http.NewRequest("GET", edgeURL+"/", nil) |
| 76 | if err != nil { |
| 77 | t.Fatalf("new request: %v", err) |
| 78 | } |
| 79 | req.Host = host |
| 80 | resp, err := http.DefaultClient.Do(req) |
| 81 | if err != nil { |
| 82 | t.Fatalf("GET %s (host %s): %v", edgeURL, host, err) |
| 83 | } |
| 84 | return resp |
| 85 | } |
| 86 | |
| 87 | // TestFleetRouteProxiesToBoundBox: rule 1 — a route bound (FleetBox) to a |
| 88 | // box whose channel is connected rides that exact channel. Bound to pi-2 |
| 89 | // DELIBERATELY, not pi-1: ConnectedOfFleet sorts ascending, so with both |
| 90 | // connected the fallback rule (rule 2) would ALSO pick "acme/pi-1" — binding |
| 91 | // there would make rule 1 and rule 2 indistinguishable (both answer "pi-1"), |
| 92 | // and this test would stay green even with the entire rule-1 branch deleted |
| 93 | // from fleetChannelFor. Binding to pi-2 (which sorts LAST) means only rule 1 |
| 94 | // can produce "pi-2" here — the fallback alone would answer "pi-1". |
| 95 | func TestFleetRouteProxiesToBoundBox(t *testing.T) { |
| 96 | edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1", "pi-2": "tok2"}, registry.Route{ |
| 97 | Host: "app.example", Kind: "instance", Status: "ready", |
| 98 | Region: registry.PlacementTarget("acme"), FleetBox: "pi-2", |
| 99 | }) |
| 100 | stop1 := startFakeRemoteHost(t, edgeURL, "acme/pi-1", "tok1", namedHandler("pi-1")) |
| 101 | defer stop1() |
| 102 | stop2 := startFakeRemoteHost(t, edgeURL, "acme/pi-2", "tok2", namedHandler("pi-2")) |
| 103 | defer stop2() |
| 104 | waitForTunnelState(t, hub, "acme/pi-1", "connected", e2eReconnectBound) |
| 105 | waitForTunnelState(t, hub, "acme/pi-2", "connected", e2eReconnectBound) |
| 106 | |
| 107 | code, body := getHost(t, edgeURL, "app.example") |
| 108 | if code != http.StatusOK || body != "pi-2" { |
| 109 | t.Fatalf("got %d %q, want 200 %q (the bound box, which sorts LAST — a fallback-only implementation would answer pi-1 instead)", code, body, "pi-2") |
| 110 | } |
| 111 | } |
| 112 | |
| 113 | // TestFleetRouteFallsOverWhenBoundBoxIsDark: rule 2 — the bound box (pi-1) |
| 114 | // never connects at all; the only connected channel of the fleet (pi-2) |
| 115 | // serves the request instead of a 502/503. |
| 116 | func TestFleetRouteFallsOverWhenBoundBoxIsDark(t *testing.T) { |
| 117 | edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1", "pi-2": "tok2"}, registry.Route{ |
| 118 | Host: "app.example", Kind: "instance", Status: "ready", |
| 119 | Region: registry.PlacementTarget("acme"), FleetBox: "pi-1", |
| 120 | }) |
| 121 | // pi-1 is never dialed — bound but dark. |
| 122 | stop2 := startFakeRemoteHost(t, edgeURL, "acme/pi-2", "tok2", namedHandler("pi-2")) |
| 123 | defer stop2() |
| 124 | waitForTunnelState(t, hub, "acme/pi-2", "connected", e2eReconnectBound) |
| 125 | |
| 126 | if hub.State("acme/pi-1") == "connected" { |
| 127 | t.Fatal("test setup: pi-1 must not be connected for this case") |
| 128 | } |
| 129 | code, body := getHost(t, edgeURL, "app.example") |
| 130 | if code != http.StatusOK || body != "pi-2" { |
| 131 | t.Fatalf("got %d %q, want 200 %q (fallover to the only connected box)", code, body, "pi-2") |
| 132 | } |
| 133 | } |
| 134 | |
| 135 | // TestFleetRouteDarkFleetAnswers503: rule 3 — no channel of the fleet is |
| 136 | // connected at all (bound or otherwise): a clean 503 + Retry-After: 2, not |
| 137 | // the tunnel path's 502. |
| 138 | func TestFleetRouteDarkFleetAnswers503(t *testing.T) { |
| 139 | edgeURL, _ := fleetTestEdge(t, map[string]string{"pi-1": "tok1"}, registry.Route{ |
| 140 | Host: "app.example", Kind: "instance", Status: "ready", |
| 141 | Region: registry.PlacementTarget("acme"), FleetBox: "pi-1", |
| 142 | }) |
| 143 | // Nothing ever dials in. |
| 144 | resp := getHostFull(t, edgeURL, "app.example") |
| 145 | defer resp.Body.Close() |
| 146 | if resp.StatusCode != http.StatusServiceUnavailable { |
| 147 | t.Fatalf("got %d, want 503", resp.StatusCode) |
| 148 | } |
| 149 | if got := resp.Header.Get("Retry-After"); got != "2" { |
| 150 | t.Fatalf("Retry-After = %q, want %q", got, "2") |
| 151 | } |
| 152 | } |
| 153 | |
| 154 | // TestFleetRouteWithNoBoundBoxUsesFallback: FleetBox can legitimately be "" |
| 155 | // (placement wants a fleet but no box has been assigned yet) — that must |
| 156 | // skip straight to the fallback rule rather than erroring on an empty bound |
| 157 | // channel name. |
| 158 | func TestFleetRouteWithNoBoundBoxUsesFallback(t *testing.T) { |
| 159 | edgeURL, hub := fleetTestEdge(t, map[string]string{"pi-1": "tok1"}, registry.Route{ |
| 160 | Host: "app.example", Kind: "instance", Status: "ready", |
| 161 | Region: registry.PlacementTarget("acme"), FleetBox: "", |
| 162 | }) |
| 163 | stop1 := startFakeRemoteHost(t, edgeURL, "acme/pi-1", "tok1", namedHandler("pi-1")) |
| 164 | defer stop1() |
| 165 | waitForTunnelState(t, hub, "acme/pi-1", "connected", e2eReconnectBound) |
| 166 | |
| 167 | code, body := getHost(t, edgeURL, "app.example") |
| 168 | if code != http.StatusOK || body != "pi-1" { |
| 169 | t.Fatalf("got %d %q, want 200 %q", code, body, "pi-1") |
| 170 | } |
| 171 | } |
| 172 | |
| 173 | // TestExplicitTunnelPrecedesFleetPlacement: a route can carry both an |
| 174 | // explicit Tunnel binding and a fleet Region ("carlos route -tunnel" doesn't |
| 175 | // clear Region, mirroring the existing Tunnel-vs-Socket/Addr precedence) — |
| 176 | // the explicit Tunnel wins, per the brief's "Explicit rt.Tunnel keeps |
| 177 | // precedence (legacy/tito path unchanged)". |
| 178 | func TestExplicitTunnelPrecedesFleetPlacement(t *testing.T) { |
| 179 | reg := openReg(t) |
| 180 | if err := reg.UpsertFleet(registry.Fleet{Name: "acme", Account: "amadan", Bucket: "acme-data"}); err != nil { |
| 181 | t.Fatal(err) |
| 182 | } |
| 183 | if err := reg.AddFleetBox(registry.FleetBox{Fleet: "acme", Label: "pi-1", TokenHash: sha256Hex("tok1"), Status: "active"}); err != nil { |
| 184 | t.Fatal(err) |
| 185 | } |
| 186 | if err := reg.Upsert(registry.Route{ |
| 187 | Host: "app.example", Kind: "instance", Status: "ready", |
| 188 | Region: registry.PlacementTarget("acme"), FleetBox: "pi-1", |
| 189 | Tunnel: "legacy-box", |
| 190 | }); err != nil { |
| 191 | t.Fatal(err) |
| 192 | } |
| 193 | e, err := New(reg, nil) |
| 194 | if err != nil { |
| 195 | t.Fatal(err) |
| 196 | } |
| 197 | srv := httptest.NewServer(e) |
| 198 | defer srv.Close() |
| 199 | u, err := url.Parse(srv.URL) |
| 200 | if err != nil { |
| 201 | t.Fatal(err) |
| 202 | } |
| 203 | hub := NewTunnelHub(u.Hostname(), map[string]string{"legacy-box": "sekrit"}) |
| 204 | e.EnableTunnels(hub) |
| 205 | if err := e.Reload(); err != nil { |
| 206 | t.Fatal(err) |
| 207 | } |
| 208 | stop := startFakeRemoteHost(t, srv.URL, "legacy-box", "sekrit", namedHandler("legacy")) |
| 209 | defer stop() |
| 210 | waitForTunnelState(t, hub, "legacy-box", "connected", e2eReconnectBound) |
| 211 | // The fleet channel (acme/pi-1) is never dialed at all — if the fleet |
| 212 | // path were somehow taken, this would 503, not serve "legacy". |
| 213 | code, body := getHost(t, srv.URL, "app.example") |
| 214 | if code != http.StatusOK || body != "legacy" { |
| 215 | t.Fatalf("got %d %q, want 200 %q (explicit Tunnel must win over fleet placement)", code, body, "legacy") |
| 216 | } |
| 217 | } |
| 218 | |
| 219 | // TestNonFleetRoutesUntouchedByFleetDispatch: an ordinary Addr-backed route |
| 220 | // with no fleet placement at all (Region == "") must be completely unaffected |
| 221 | // by the new dispatch case — same behavior as before Task 4. |
| 222 | func TestNonFleetRoutesUntouchedByFleetDispatch(t *testing.T) { |
| 223 | backend, _ := hostEchoBackend(t, "plain backend") |
| 224 | reg := openReg(t) |
| 225 | if err := reg.Upsert(registry.Route{ |
| 226 | Host: "plain.example", Kind: "service", Status: "ready", |
| 227 | Addr: backend.Listener.Addr().String(), |
| 228 | }); err != nil { |
| 229 | t.Fatal(err) |
| 230 | } |
| 231 | e, err := New(reg, nil) |
| 232 | if err != nil { |
| 233 | t.Fatal(err) |
| 234 | } |
| 235 | srv := httptest.NewServer(e) |
| 236 | defer srv.Close() |
| 237 | code, body := getHost(t, srv.URL, "plain.example") |
| 238 | if code != http.StatusOK || body != "plain backend" { |
| 239 | t.Fatalf("got %d %q, want 200 %q", code, body, "plain backend") |
| 240 | } |
| 241 | } |
| 242 | |
| 243 | // TestFleetRouteIsNeverActivatorManaged mirrors |
| 244 | // TestTunnelRouteIsNeverActivatorManaged (tunnel_test.go) for the fleet |
| 245 | // dispatch case: a fleet-placed, Hibernate:true route must never be |
| 246 | // activator-managed on THIS box — whichever box ends up serving it wakes its |
| 247 | // OWN local instance, so a platform-side wake here would start a second copy |
| 248 | // beside it. Today this rests on viaTunnel being set for the fleet case too |
| 249 | // (edge.go); this test is the regression pin so a future reorder or refactor |
| 250 | // of that switch can't silently drop it — the exclusion would otherwise only |
| 251 | // be caught by the tunnel-specific test above, which a fleet-only regression |
| 252 | // wouldn't touch at all. |
| 253 | func TestFleetRouteIsNeverActivatorManaged(t *testing.T) { |
| 254 | reg := openReg(t) |
| 255 | if err := reg.Upsert(registry.Route{ |
| 256 | Host: "fleet-hib.example", Kind: "instance", Status: "ready", |
| 257 | Region: registry.PlacementTarget("acme"), FleetBox: "pi-1", Hibernate: true, |
| 258 | }); err != nil { |
| 259 | t.Fatal(err) |
| 260 | } |
| 261 | act := &fakeActivator{} |
| 262 | e, err := New(reg, act) |
| 263 | if err != nil { |
| 264 | t.Fatal(err) |
| 265 | } |
| 266 | e.EnableTunnels(NewTunnelHub("tunnel.test", nil)) |
| 267 | if err := e.Reload(); err != nil { |
| 268 | t.Fatal(err) |
| 269 | } |
| 270 | if last := act.retired[len(act.retired)-1]; last["fleet-hib.example"] { |
| 271 | t.Fatalf("fleet-served host kept in the activator's managed set: %v", last) |
| 272 | } |
| 273 | req := httptest.NewRequest("GET", "http://fleet-hib.example/", nil) |
| 274 | req.Host = "fleet-hib.example" |
| 275 | e.ServeHTTP(httptest.NewRecorder(), req) |
| 276 | if act.served != 0 { |
| 277 | t.Fatalf("activator woke a local process for a fleet-served host (%d call(s))", act.served) |
| 278 | } |
| 279 | } |
| 280 | |
| 281 | // TestConnectedOfFleetSortedAndFiltered pins ConnectedOfFleet directly: only |
| 282 | // names under "<fleet>/" come back, sorted, and a same-prefixed different |
| 283 | // fleet ("acme2") is never mistaken for a member of "acme". |
| 284 | func TestConnectedOfFleetSortedAndFiltered(t *testing.T) { |
| 285 | hub := NewTunnelHub("tunnel.test", nil) |
| 286 | for _, name := range []string{"acme/pi-2", "acme/pi-1", "acme2/pi-1", "other/pi-1"} { |
| 287 | c1, c2 := net.Pipe() |
| 288 | t.Cleanup(func() { c2.Close() }) |
| 289 | hub.mu.Lock() |
| 290 | hub.conns[name] = &tunnelConn{raw: c1} |
| 291 | hub.mu.Unlock() |
| 292 | } |
| 293 | got := hub.ConnectedOfFleet("acme") |
| 294 | want := []string{"acme/pi-1", "acme/pi-2"} |
| 295 | if len(got) != len(want) || got[0] != want[0] || got[1] != want[1] { |
| 296 | t.Fatalf("ConnectedOfFleet(acme) = %v, want %v", got, want) |
| 297 | } |
| 298 | } |
| 299 | |
amadan