amadan

carlos / platform Public

edge: multi-upstream addr proxy — round-robin live set, in-request dial retry, 503+Retry-After when dark

Reload's addr arm now gates on Route.AddrList() rather than Route.Addr, so a
route carrying the Addrs column is served, and picks its proxy by whether the
route is PROBED, not by how many upstreams it names:

  - no health config -> tcpProxy on addrs[0], byte-identical to the old path
    (a dead backend still answers the stdlib 502).
  - health config -> multiProxy, single upstream or five: a probed route is on
    the dashboard and answers 503 + Retry-After: 2 when it is dark.

An unparseable health column costs the route its probing, never its service.

multiProxy picks its upstream in DialContext — the last moment at which
nothing has been sent — so a refused box is ejected from the live set and the
next one tried inside the same request, with no user-visible error and no
replayed body. Per-route round-robin cursor, 3s per ATTEMPT (not per request),
and no eject when it is our own context that died. Upstream keep-alives are
off: the stdlib pools by the outbound authority, which is a fixed placeholder
here, so a pooled connection would pin the route to one upstream and outlive
an Eject.

Probe specs are collected on the same Reload pass and Sync'd BEFORE the
upstream map swap — the opposite order to RetireExcept, because a prober that
does not exist yet is indistinguishable from one whose upstreams are all down,
and a newly added route must never 503 for the gap.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell a4e2e7f56c596d29c8e94c53c66e923a5fdf448a parent e9b7779
2 files changed, +622 −3
  • internal/edge/edge.go +230 −3
  • internal/edge/edge_test.go +392 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 219e8697e..97a411db4 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -20,6 +20,7 @@ import (
"bufio"
"context"
"crypto/ed25519"
+ "encoding/json"
"errors"
"flag"
"fmt"
@@ -42,6 +43,7 @@ import (
"golang.org/x/crypto/acme/autocert"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -139,6 +141,14 @@ type Edge struct {
// because registration happens in RunWithActivator after Watch may
// already be reloading.
reloadHook atomic.Pointer[func()]
+ // probers owns the health-probe goroutines for multi-upstream addr routes
+ // (prober.go) and the live set multiProxy picks from. Created in New,
+ // BEFORE the startup Reload — Reload calls Sync on it unconditionally and
+ // a nil *proberSet panics by design, so a construction path that skipped
+ // this would be a wiring bug that fails loudly rather than a fleet that
+ // quietly 503s. Never stopped in production (the edge lives as long as the
+ // process); tests Stop it so probe goroutines don't outlive them.
+ probers *proberSet
}
// upstream is one served host: the handler for its backend, plus — when the
@@ -155,7 +165,7 @@ type upstream struct {
// New builds an Edge over reg and performs the startup reload. act may be nil
// (no hibernation).
func New(reg *registry.Registry, act Activator) (*Edge, error) {
- e := &Edge{reg: reg, act: act}
+ e := &Edge{reg: reg, act: act, probers: newProberSet()}
if err := e.Reload(); err != nil {
return nil, err
}
@@ -275,6 +285,11 @@ func (e *Edge) Reload() error {
// The hosts the activator should manage after this reload — feeds both the
// per-upstream gate and the retire pass below.
managed := make(map[string]bool)
+ // The probing jobs this reload's route set implies — one entry per addr
+ // route that carries a health config, fed to proberSet.Sync below. Hosts
+ // absent from it have their probers stopped, exactly as RetireExcept
+ // retires activator children for hosts absent from `managed`.
+ probeSpecs := make(map[string]probeSpec)
for _, t := range routes {
// Only serve ready routes: a 'creating'/'failed' entry has no working
// backend, so proxying it would just stall → 502 — AND, since Allowed
@@ -350,8 +365,53 @@ func (e *Edge) Reload() error {
p := unixProxy(t.Socket)
p.ModifyResponse = e.purgeModifier(t)
proxy = p
- case t.Addr != "":
- p := tcpProxy(t.Addr)
+ // TCP upstreams. The gate is AddrList (registry), not t.Addr, so a
+ // multi-upstream route — which writes the Addrs column and leaves the
+ // legacy Addr empty — is served at all; AddrList folds the legacy
+ // single Addr into a one-element list, so nothing about the old shape
+ // moves. Position in the dispatch order is unchanged (tunnel > socket >
+ // addrs > static > status).
+ //
+ // Health is what selects the proxy, not the addr COUNT:
+ //
+ // - no health config → tcpProxy, byte-identical to the pre-#152
+ // path. An operator who never asked for probing gets exactly the
+ // behavior they had, including the stdlib 502 on a dead backend.
+ // - health config → multiProxy, whether the route names one upstream
+ // or five. A single PROBED upstream is on the dashboard and answers
+ // 503 + Retry-After when it is down, which is the point of asking
+ // for a probe on a one-box route.
+ //
+ // A route with several addrs but no health config takes the unprobed
+ // path on addrs[0]: without a probe there is no live set to balance
+ // over, and inventing one would mean load-balancing onto upstreams
+ // nothing is checking. It is a config gap, not a serving failure —
+ // mint always writes both columns together (provision), so this is
+ // only reachable via a hand-written row.
+ case len(t.AddrList()) > 0:
+ addrs := t.AddrList()
+ var p *httputil.ReverseProxy
+ switch {
+ case t.Health == "":
+ p = tcpProxy(addrs[0])
+ default:
+ var h provision.Health
+ if err := json.Unmarshal([]byte(t.Health), &h); err != nil {
+ // Never a dead route: a health column we cannot parse costs
+ // the route its probing, not its service. Logged every
+ // reload on purpose — it is a config error someone has to
+ // fix, and the route silently losing HA is worse than the
+ // log line.
+ log.Printf("edge: route %s: unparseable health config (%v) — serving %s unprobed", t.Host, err, addrs[0])
+ p = tcpProxy(addrs[0])
+ break
+ }
+ // Raw parse, deliberately un-normalized: Sync runs every spec
+ // through Health.WithDefaults itself, and normalizing here too
+ // would just be a second place to keep in step with it.
+ probeSpecs[t.Host] = probeSpec{Host: t.Host, Addrs: addrs, Health: h}
+ p = multiProxy(t.Host, e.probers)
+ }
p.ModifyResponse = e.purgeModifier(t)
proxy = p
case t.Kind == "static":
@@ -451,6 +511,25 @@ func (e *Edge) Reload() error {
}
e.fleetAccounts.Store(&accounts)
}
+ // Probers BEFORE the map swap — the opposite order to RetireExcept below,
+ // and deliberately so. multiProxy answers 503 + Retry-After when the
+ // prober has nothing live for its host, and a prober this Sync has not
+ // installed yet is indistinguishable from one whose upstreams are all
+ // down. Swapping first would therefore open a window — however short —
+ // in which a request for a newly added (or newly re-specced) multi-upstream
+ // route reaches a handler whose live set is empty and gets a 503 for a
+ // perfectly healthy fleet. Syncing first cannot cause the mirror-image
+ // problem, because a prober is optimistic at birth (prober.go rule 1): the
+ // new host's upstreams are all live the instant Sync returns, before any
+ // request can reach the handler that reads them.
+ //
+ // The reverse case — a route REMOVED by this reload — loses its prober
+ // before its handler, so a request landing in that window sees an empty
+ // live set and gets 503 instead of proxying a route that is about to stop
+ // existing. That is the trade, and it is the right way round: a 503 for a
+ // host being deleted is a moment early, a 503 for a host being added is
+ // simply wrong.
+ e.probers.Sync(probeSpecs)
e.upstreams.Store(&m)
// Reconcile the activator with the new route set: a host un-flagged (or
// removed) since the last reload may still have a live child holding its
@@ -1082,6 +1161,154 @@ func tcpProxy(addr string) *httputil.ReverseProxy {
}
}
+// errNoLiveUpstream is the sentinel multiProxy's dialer returns when a probed
+// route has nothing left to try — an empty live set, or every live upstream
+// refusing the dial inside this one request. multiProxy's ErrorHandler turns
+// it into 503 + Retry-After: 2, the same "nobody is home right now, come back
+// shortly" answer errFleetDark gets, rather than the ReverseProxy default's
+// 502 (which reads as "your backend is broken" and is what an UNPROBED addr
+// route still correctly returns).
+//
+// It never escapes the route it came from: each multiProxy has its own
+// ErrorHandler, and the only thing that can produce this error is that
+// proxy's own DialContext.
+var errNoLiveUpstream = errors.New("edge: no live upstream")
+
+// multiDialHost is the placeholder authority multiProxy's Rewrite puts on the
+// outbound URL. Nothing ever resolves it: DialContext ignores the address
+// entirely and dials the live set instead. `.invalid` is the reserved TLD for
+// exactly this (RFC 2606), so if it ever DID reach a resolver — a wiring
+// mistake — it fails immediately and visibly rather than hitting somebody's
+// wildcard DNS.
+const multiDialHost = "multi.invalid"
+
+// multiDialTimeout bounds ONE dial attempt. Per attempt, not per request: a
+// route with three upstreams whose first two are black-holed must still reach
+// the third, and a single request-wide budget would spend it all on the first
+// dead box. Three seconds is long enough for a loaded box on a real network to
+// complete a handshake and short enough that a fully dark route answers its
+// 503 well inside any sane client timeout.
+const multiDialTimeout = 3 * time.Second
+
+// multiProxy reverse-proxies a probed addr route across its live upstreams
+// (prober.go). Same Host preservation and purge-header discipline as tcpProxy;
+// what it adds is where the bytes go.
+//
+// It takes the host and not the route's addr list because the addr list is not
+// its input: the prober owns it (Reload hands the same list to Sync), and the
+// only set this proxy may dial is the LIVE one. Closing over the configured
+// addrs as well would create a second, staler copy of the route's upstreams
+// and invite exactly the fallback the design forbids — dialing a box the probe
+// has already declared dead.
+//
+// Selection happens in DialContext, not in Rewrite, and that is the whole
+// design. A dial is the last moment at which NOTHING has been sent: the
+// request body is untouched and no response has begun, so a refused upstream
+// can be ejected and the next one tried inside the same request, with the
+// client none the wiser. Retrying one layer up (at RoundTrip) would mean
+// replaying a request that may already have had its body consumed, which is
+// only safe for a subset of requests — this way it is safe for all of them.
+//
+// The live set is read PER REQUEST rather than captured, mirroring
+// fleetProxy/tunnelProxy: an upstream that fails away (or rejoins) between two
+// requests is picked up with no Reload needed.
+//
+// Keep-alives to upstreams are off, which is a real cost paid for two real
+// properties. The stdlib pools connections by the outbound URL's authority,
+// and every request through this proxy carries the same placeholder one — so a
+// pooled connection would be reused regardless of WHICH upstream it goes to.
+// That would (a) pin a route to whichever upstream happened to answer first,
+// making the round-robin below decorative, and (b) survive an Eject: the
+// upstream the prober just took out of rotation would keep serving traffic
+// over the connection already in the pool. Correct failover is worth a TCP
+// handshake per request to a backend that is, in every deployment this exists
+// for, on the same network. The unprobed tcpProxy path keeps its pool
+// untouched.
+func multiProxy(host string, p *proberSet) *httputil.ReverseProxy {
+ // Per-ROUTE cursor: closed over by this proxy alone, so two routes cannot
+ // interleave their rotations (a global counter would make a two-upstream
+ // route's balance depend on the request rate of every other route on the
+ // box). Reload builds a new proxy, so the cursor resets with the route's
+ // upstream set — which is what you want, since the set it was counting
+ // over no longer exists.
+ var cursor atomic.Uint64
+ dialer := &net.Dialer{Timeout: multiDialTimeout}
+ return &httputil.ReverseProxy{
+ Rewrite: func(pr *httputil.ProxyRequest) {
+ pr.SetXForwarded()
+ pr.Out.URL.Scheme = "http"
+ pr.Out.URL.Host = multiDialHost // placeholder; the dialer below ignores it
+ pr.Out.Host = pr.In.Host // backend sees the real hostname
+ scrubInboundPurgeHeader(pr)
+ },
+ Transport: &http.Transport{
+ DialContext: func(ctx context.Context, network, _ string) (net.Conn, error) {
+ live := p.Live(host)
+ if len(live) == 0 {
+ return nil, errNoLiveUpstream
+ }
+ // Start where the last request left off, then walk the whole
+ // snapshot once. Snapshot rather than re-reading Live per
+ // attempt: each Eject below shrinks the live set, and re-reading
+ // would skip untried upstreams as the list moved under us.
+ //
+ // The index arithmetic stays in uint64 all the way to the
+ // modulo. Converting the counter to int first would wrap
+ // NEGATIVE once it passed 2^31 on a 32-bit box (the Pi fleet),
+ // and Go's modulo keeps the sign — so the very next request
+ // after that wrap would index out of range and panic the edge.
+ start := cursor.Add(1) - 1
+ n := uint64(len(live))
+ var lastErr error
+ for i := range live {
+ if err := ctx.Err(); err != nil {
+ // The client went away (or the request deadline
+ // expired) mid-loop. Stop, and do NOT eject: a dial we
+ // abandoned says nothing about the upstream's health,
+ // and marking it down here would take a healthy box out
+ // of rotation because one user hit stop.
+ return nil, err
+ }
+ addr := live[(start+uint64(i))%n]
+ conn, err := dialer.DialContext(ctx, network, addr)
+ if err == nil {
+ return conn, nil
+ }
+ lastErr = err
+ if ctx.Err() != nil {
+ return nil, err // as above: our own cancellation, not the upstream's fault
+ }
+ // The fast path the probe cadence cannot match: a box that
+ // died since the last tick is out of rotation now, for
+ // every other in-flight request too, instead of collecting
+ // failures until fail_after.
+ p.Eject(host, addr)
+ }
+ return nil, fmt.Errorf("%w for %s: %d upstream(s) refused, last: %v",
+ errNoLiveUpstream, host, len(live), lastErr)
+ },
+ // No MaxIdleConns/IdleConnTimeout twin of tcpProxy's: with
+ // keep-alives off nothing is ever pooled, so the idle bounds that
+ // stop a replaced proxy leaking pooled connections have nothing to
+ // bound. Every connection this transport opens belongs to exactly
+ // one request and closes with it.
+ DisableKeepAlives: true, // see the pooling note above
+ },
+ ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
+ if errors.Is(err, errNoLiveUpstream) {
+ w.Header().Set("Retry-After", "2")
+ http.Error(w, "no healthy upstream", http.StatusServiceUnavailable)
+ return
+ }
+ // Anything else is a real proxy failure (upstream hung up
+ // mid-response, malformed reply): same shape as
+ // httputil.ReverseProxy's own default handler, log and 502.
+ log.Printf("edge: multi-upstream %s proxy error: %v", host, err)
+ w.WriteHeader(http.StatusBadGateway)
+ },
+ }
+}
+
// tunnelProxy reverse-proxies over a named reverse channel (TunnelHub).
// Same Host preservation and purge-header discipline as tcpProxy — a
// tunnel-backed host is exactly as untrusted as an addr-backed one. The
diff --git a/internal/edge/edge_test.go b/internal/edge/edge_test.go
index 80d00b007..4ada76d02 100644
--- a/internal/edge/edge_test.go
+++ b/internal/edge/edge_test.go
@@ -20,6 +20,7 @@ import (
"testing"
"time"
+ "github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
)
@@ -1527,3 +1528,394 @@ func TestNoVersionHeaderOnUnversionedRoute(t *testing.T) {
t.Fatalf("X-Carlos-Version present (%v), want absent for an unversioned route", got)
}
}
+
+// ---------------------------------------------------------------------------
+// Multi-upstream addr routes (HA design Part II, issue #152)
+// ---------------------------------------------------------------------------
+
+// testProbePath is the health path every multi-upstream test configures. It is
+// deliberately NOT "/": countingBackend ignores it, so a counter can never
+// confuse the prober's own traffic for proxied requests — a test that counted
+// probes would report "both upstreams were used" on a proxy that balanced
+// nothing at all.
+const testProbePath = "/healthz"
+
+// countingBackend is an upstream that counts the PROXIED requests it served
+// and answers a fixed body. Probe requests (testProbePath) are answered 200
+// but never counted.
+func countingBackend(t *testing.T, body string) (string, *atomic.Int64) {
+ t.Helper()
+ hits := new(atomic.Int64)
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == testProbePath {
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+ hits.Add(1)
+ io.WriteString(w, body)
+ }))
+ t.Cleanup(ts.Close)
+ return ts.Listener.Addr().String(), hits
+}
+
+// deadAddr reserves a real loopback port and closes the listener immediately,
+// yielding an address that is well-formed and guaranteed to REFUSE
+// connections. That refusal is instant, so a test asserting the dial-failure
+// path never waits out a timeout — and, unlike a made-up port, it cannot
+// collide with something else the machine is running.
+func deadAddr(t *testing.T) string {
+ t.Helper()
+ ln, err := net.Listen("tcp", "127.0.0.1:0")
+ if err != nil {
+ t.Fatalf("listen: %v", err)
+ }
+ addr := ln.Addr().String()
+ if err := ln.Close(); err != nil {
+ t.Fatalf("close reserved listener: %v", err)
+ }
+ return addr
+}
+
+// healthJSON marshals a probe config the way provision does when it mints a
+// route, so the tests exercise the same column bytes Reload parses in
+// production.
+func healthJSON(t *testing.T, h provision.Health) string {
+ t.Helper()
+ b, err := json.Marshal(h)
+ if err != nil {
+ t.Fatalf("marshal health: %v", err)
+ }
+ return string(b)
+}
+
+// frozenHealth is a probe config whose cadence cannot fire within a test: one
+// tick runs immediately on Sync (hostProber.run), then nothing for a minute,
+// and fail_after is far beyond what a single tick could reach. Any live-set
+// change a test observes under it therefore came from the REQUEST path's
+// dial-failure fast path (proberSet.Eject), never from a probe.
+func frozenHealth() provision.Health {
+ return provision.Health{Path: testProbePath, Status: 200, IntervalMS: 60000, FailAfter: 1000, RiseAfter: 1000}
+}
+
+func TestMultiUpstreamBalancesAcrossLiveBackends(t *testing.T) {
+ addrA, hitsA := countingBackend(t, "A")
+ addrB, hitsB := countingBackend(t, "B")
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "ha.example.com", Kind: "service", Status: "ready",
+ Addrs: addrA + "," + addrB,
+ Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ for i := 0; i < 20; i++ {
+ code, body := getHost(t, srv.URL, "ha.example.com")
+ if code != http.StatusOK {
+ t.Fatalf("request %d: status = %d, want 200", i, code)
+ }
+ if body != "A" && body != "B" {
+ t.Fatalf("request %d: body = %q, want one of the two upstreams", i, body)
+ }
+ }
+ if hitsA.Load() == 0 || hitsB.Load() == 0 {
+ t.Fatalf("upstream hits = A:%d B:%d, want BOTH non-zero (round-robin over the live set)",
+ hitsA.Load(), hitsB.Load())
+ }
+ if total := hitsA.Load() + hitsB.Load(); total != 20 {
+ t.Fatalf("upstreams served %d requests, want exactly 20 (no request served twice or dropped)", total)
+ }
+}
+
+func TestDialFailureRetriesOtherUpstreamNoUserError(t *testing.T) {
+ dead := deadAddr(t)
+ addrA, hitsA := countingBackend(t, "A")
+
+ reg := openReg(t)
+ // Dead upstream FIRST in the list, so the very first request must retry
+ // past it inside the same request rather than answering an error.
+ if err := reg.Upsert(registry.Route{
+ Host: "ha.example.com", Kind: "service", Status: "ready",
+ Addrs: dead + "," + addrA,
+ Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ for i := 0; i < 10; i++ {
+ code, body := getHost(t, srv.URL, "ha.example.com")
+ if code != http.StatusOK {
+ t.Fatalf("request %d: status = %d, want 200 — a dead upstream must cost a retry, not a user-visible error", i, code)
+ }
+ if body != "A" {
+ t.Fatalf("request %d: body = %q, want %q (the live upstream)", i, body, "A")
+ }
+ }
+ if got := hitsA.Load(); got != 10 {
+ t.Fatalf("live upstream served %d requests, want 10", got)
+ }
+ // The fast path: frozenHealth's cadence cannot have ejected anything, so
+ // this can only be the in-request dial failure.
+ if live := e.probers.Live("ha.example.com"); !slices.Equal(live, []string{addrA}) {
+ t.Fatalf("live set = %v, want [%s] — a failed dial must eject %s immediately", live, addrA, dead)
+ }
+}
+
+func TestAllUpstreamsDownAnswers503WithRetryAfter(t *testing.T) {
+ dead1, dead2 := deadAddr(t), deadAddr(t)
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "ha.example.com", Kind: "service", Status: "ready",
+ Addrs: dead1 + "," + dead2,
+ Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ resp := getHostHeader(t, srv.URL, "ha.example.com")
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d, want 503 (a dark upstream set is 'come back shortly', not the stdlib's bare 502)", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Retry-After"); got != "2" {
+ t.Fatalf("Retry-After = %q, want %q", got, "2")
+ }
+ // A second request must answer the same way rather than hanging or
+ // escalating: every upstream is ejected by now, so the live set is empty
+ // and the transport never dials at all.
+ resp2 := getHostHeader(t, srv.URL, "ha.example.com")
+ if resp2.StatusCode != http.StatusServiceUnavailable || resp2.Header.Get("Retry-After") != "2" {
+ t.Fatalf("second request = %d Retry-After %q, want 503/2 with an empty live set",
+ resp2.StatusCode, resp2.Header.Get("Retry-After"))
+ }
+ if live := e.probers.Live("ha.example.com"); len(live) != 0 {
+ t.Fatalf("live set = %v, want empty after both upstreams refused", live)
+ }
+}
+
+func TestSingleAddrWithHealthGets503Semantics(t *testing.T) {
+ dead := deadAddr(t)
+
+ reg := openReg(t)
+ // One upstream, carried in the LEGACY Addr column, but with health
+ // configured: probing is what buys the 503 semantics, not the count.
+ if err := reg.Upsert(registry.Route{
+ Host: "solo.example.com", Kind: "service", Status: "ready",
+ Addr: dead,
+ Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ resp := getHostHeader(t, srv.URL, "solo.example.com")
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d, want 503 for a probed single upstream", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Retry-After"); got != "2" {
+ t.Fatalf("Retry-After = %q, want %q", got, "2")
+ }
+ // A probed single-addr route is on the dashboard like any other.
+ if st, ok := e.probers.State()["solo.example.com"]; !ok || len(st) != 1 || st[0].Addr != dead {
+ t.Fatalf("prober state = %v, want one entry for %s", st, dead)
+ }
+}
+
+func TestLegacySingleAddrRouteUnchanged(t *testing.T) {
+ dead := deadAddr(t)
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "legacy.example.com", Kind: "service", Status: "ready", Addr: dead,
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ resp := getHostHeader(t, srv.URL, "legacy.example.com")
+ if resp.StatusCode != http.StatusBadGateway {
+ t.Fatalf("status = %d, want 502 — an UNPROBED route must keep its pre-multi-upstream behavior byte for byte", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Retry-After"); got != "" {
+ t.Fatalf("Retry-After = %q, want absent on the legacy path", got)
+ }
+ if st := e.probers.State(); len(st) != 0 {
+ t.Fatalf("prober state = %v, want empty — a route with no health config is not probed", st)
+ }
+}
+
+func TestReloadSwapsUpstreamSet(t *testing.T) {
+ addrA, _ := countingBackend(t, "A")
+ addrB, _ := countingBackend(t, "B")
+ health := healthJSON(t, frozenHealth())
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "ha.example.com", Kind: "service", Status: "ready",
+ Addrs: addrA, Health: health,
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ if code, body := getHost(t, srv.URL, "ha.example.com"); code != http.StatusOK || body != "A" {
+ t.Fatalf("pre-reload = %d %q, want 200 %q", code, body, "A")
+ }
+
+ // Write via a SECOND handle, the cross-process shape a reconcile pass has.
+ reg2, err := registry.Open(reg.Path())
+ if err != nil {
+ t.Fatalf("open second handle: %v", err)
+ }
+ if err := reg2.SetUpstreams("ha.example.com", addrB, health); err != nil {
+ t.Fatalf("SetUpstreams: %v", err)
+ }
+ if err := e.Reload(); err != nil {
+ t.Fatalf("Reload: %v", err)
+ }
+
+ if code, body := getHost(t, srv.URL, "ha.example.com"); code != http.StatusOK || body != "B" {
+ t.Fatalf("post-reload = %d %q, want 200 %q (the new upstream set must be served)", code, body, "B")
+ }
+ st := e.probers.State()["ha.example.com"]
+ if len(st) != 1 || st[0].Addr != addrB {
+ t.Fatalf("prober state = %v, want exactly one entry for %s — the probe specs must follow the route", st, addrB)
+ }
+}
+
+// upgradeEchoBackend answers a protocol-switch handshake the way a WebSocket
+// server does — 101, then raw bytes on the hijacked connection — prefixing
+// whatever it reads with tag so a test can tell WHICH upstream served it.
+func upgradeEchoBackend(t *testing.T, tag string) string {
+ t.Helper()
+ ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == testProbePath {
+ w.WriteHeader(http.StatusOK)
+ return
+ }
+ hj, ok := w.(http.Hijacker)
+ if !ok {
+ http.Error(w, "not hijackable", http.StatusInternalServerError)
+ return
+ }
+ conn, buf, err := hj.Hijack()
+ if err != nil {
+ return
+ }
+ defer conn.Close()
+ buf.WriteString("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\r\n")
+ if err := buf.Flush(); err != nil {
+ return
+ }
+ line, err := buf.ReadString('\n')
+ if err != nil {
+ return
+ }
+ buf.WriteString(tag + ":" + line)
+ buf.Flush()
+ }))
+ t.Cleanup(ts.Close)
+ return ts.Listener.Addr().String()
+}
+
+// TestMultiUpstreamServesWebSocketUpgrade guards the one thing multiProxy's
+// upstream keep-alive suppression could plausibly have broken. A transport
+// with DisableKeepAlives normally adds `Connection: close` to its outbound
+// request, which would poison a protocol-switch handshake — the stdlib
+// exempts upgrade requests from that (transport.go's isProtocolSwitchHeader
+// check), and this test is what keeps that exemption load-bearing rather than
+// assumed.
+func TestMultiUpstreamServesWebSocketUpgrade(t *testing.T) {
+ addr := upgradeEchoBackend(t, "ws")
+
+ reg := openReg(t)
+ if err := reg.Upsert(registry.Route{
+ Host: "ws.example.com", Kind: "service", Status: "ready",
+ Addrs: addr, Health: healthJSON(t, frozenHealth()),
+ }); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatalf("New: %v", err)
+ }
+ defer e.probers.Stop()
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ c, err := net.Dial("tcp", strings.TrimPrefix(srv.URL, "http://"))
+ if err != nil {
+ t.Fatalf("dial edge: %v", err)
+ }
+ defer c.Close()
+ if err := c.SetDeadline(time.Now().Add(10 * time.Second)); err != nil {
+ t.Fatalf("set deadline: %v", err)
+ }
+ if _, err := fmt.Fprint(c, "GET /socket HTTP/1.1\r\nHost: ws.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n\r\n"); err != nil {
+ t.Fatalf("write handshake: %v", err)
+ }
+ br := bufio.NewReader(c)
+ resp, err := http.ReadResponse(br, nil)
+ if err != nil {
+ t.Fatalf("read handshake response: %v", err)
+ }
+ if resp.StatusCode != http.StatusSwitchingProtocols {
+ t.Fatalf("handshake status = %d, want 101 — an upgrade over a multi-upstream route must switch protocols", resp.StatusCode)
+ }
+ if _, err := fmt.Fprint(c, "ping\n"); err != nil {
+ t.Fatalf("write post-upgrade frame: %v", err)
+ }
+ line, err := br.ReadString('\n')
+ if err != nil {
+ t.Fatalf("read post-upgrade frame: %v", err)
+ }
+ if line != "ws:ping\n" {
+ t.Fatalf("post-upgrade echo = %q, want %q — bytes must flow both ways after the switch", line, "ws:ping\n")
+ }
+}