amadan

carlos / platform Public

edge: cap concurrent backhauls per home box, and refuse at the cap

Containment for the cross-region path: at most CARLOS_BACKHAUL_MAX_INFLIGHT
(default 64) backhauled requests may be in flight toward any ONE home box.
At the limit the SENDING edge answers 503 + Retry-After: 2 itself — no
dial, no cross-region bytes — and logs on the throttled per-home-box key
"backhaul-saturated "+box, never per host: a catch-all row in another
box's edges file makes every invented subdomain a target, and a
client-keyed bucket is the memory growth vector this package already
refuses elsewhere.

A concurrency bound rather than a rate, because egress is (concurrent
streams x per-stream bandwidth): a rate cap would neither bound a handful
of long-lived streams — a hop carries WebSockets and has no whole-body
deadline by design — nor tolerate a legitimate burst. A slot is held for
the life of the hop's response, streams and upgrades included, because
that IS the sustained cross-region egress being bounded.

Sender-side because every backhauled byte crosses both legs, so one cap
bounds both boxes at once; the home box needs no twin, its inbound hops
being bounded by the sum of its siblings' caps. Unset, unparseable and
non-positive values all mean the default, and there is deliberately no
"off": an unbounded unruled path is what this closes.

Dormant on a single-box deployment, where Home() never offers a target.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 8311c1d52c067595d09fe381f65ad59d38b4c12b parent 8f5d671
2 files changed, +392 −0
  • internal/edge/edge.go +96 −0
  • internal/edge/homemap_test.go +296 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 6adae2809..bfd0fda66 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -194,6 +194,13 @@ type Edge struct {
// backhaul target, and keying per host would let a stranger's stream of
// invented hostnames grow this map without limit.
backhaulTransports sync.Map
+ // backhaulInflight counts the hops currently in flight toward each home
+ // box — the containment half of the egress ruling (backhaulAcquire).
+ // Keyed by BOX for the same reason backhaulTransports is keyed by EIP: a
+ // catch-all row in another box's edges file makes every invented
+ // subdomain a backhaul target, so a per-host key would let a stranger
+ // grow this map without limit. Bounded by the deployment's box count.
+ backhaulInflight sync.Map // box → *atomic.Int64
}
// RunOption configures the Edge that Run/RunWithActivator* builds, before it
@@ -1796,6 +1803,71 @@ func fleetProxy(fleet, boxLabel string, hub *TunnelHub) *httputil.ReverseProxy {
// every request and this whole path is dead code until a second box publishes
// its edges file.
+// backhaulMaxInflight is the containment ruling of 2026-08-24: at most this
+// many backhauled requests may be in flight toward any ONE home box at a
+// time, counted on the SENDING edge. At the limit the sender answers 503 +
+// Retry-After: 2 (backhaul, below) rather than dialling.
+//
+// A concurrency bound and not a rate: cross-region egress is (concurrent
+// streams × per-stream bandwidth), so a request-rate cap would neither bound
+// a handful of long-lived streams — a hop carries WebSockets and has no
+// whole-body deadline by design — nor tolerate a legitimate burst. This
+// degrades the other way round: short requests flow at full rate forever, and
+// only slow, sustained streams pile into the cap.
+//
+// Sender-side because every backhauled byte traverses BOTH legs (home →
+// sender, sender → client), so one cap here bounds both boxes' egress at
+// once; the home box needs no twin, its inbound hops being bounded by the sum
+// of its siblings' caps.
+//
+// A var, not a const, so a test can shorten it — the same reason backhaulPort
+// is one. There is deliberately no "off" value: an unbounded cross-region
+// path is the thing this closes.
+var backhaulMaxInflight = defaultBackhaulMaxInflight
+
+const defaultBackhaulMaxInflight = 64
+
+// backhaulInflightFromEnv reads CARLOS_BACKHAUL_MAX_INFLIGHT. Unset,
+// unparseable or ≤ 0 all mean the default: a typo'd knob must not silently
+// uncap (or, worse, zero-cap) a deployment's backhaul.
+func backhaulInflightFromEnv() int {
+ raw := strings.TrimSpace(os.Getenv("CARLOS_BACKHAUL_MAX_INFLIGHT"))
+ if raw == "" {
+ return defaultBackhaulMaxInflight
+ }
+ n, err := strconv.Atoi(raw)
+ if err != nil || n <= 0 {
+ log.Printf("edge: CARLOS_BACKHAUL_MAX_INFLIGHT=%q is not a positive integer — using the default of %d concurrent backhauled requests per home box", raw, defaultBackhaulMaxInflight)
+ return defaultBackhaulMaxInflight
+ }
+ return n
+}
+
+// backhaulAcquire reserves one in-flight slot toward box, or reports
+// saturation. release must be called exactly once when the hop's ServeHTTP
+// returns — a streaming or upgraded response holds its slot for the life of
+// the stream, deliberately: a long-lived stream IS sustained cross-region
+// egress, which is the thing being bounded.
+//
+// The counter is per home BOX and the map is never keyed by anything a
+// stranger controls (backhaulInflight's own doc). Over-admission by a hair is
+// impossible and under-admission is not attempted: the count is incremented
+// first and rolled back on refusal, so two racing requests at the limit both
+// see the excess and one of them refuses.
+func (e *Edge) backhaulAcquire(box string) (release func(), ok bool) {
+ v, _ := e.backhaulInflight.LoadOrStore(box, new(atomic.Int64))
+ n := v.(*atomic.Int64)
+ if n.Add(1) > int64(backhaulMaxInflight) {
+ n.Add(-1)
+ return nil, false
+ }
+ // Once-guarded because a slot released twice would let the cap drift
+ // upwards for the life of the process — the one failure mode of a
+ // hand-released semaphore, and cheaper to make impossible than to audit.
+ var once sync.Once
+ return func() { once.Do(func() { n.Add(-1) }) }, true
+}
+
// backhaulPort is the port a home edge answers on. Always 443 in production —
// an edges file publishes a bare EIP (Task 1) and every edge's front door is
// 443 — and a var only so a test can point the dial at an httptest server, the
@@ -1863,6 +1935,25 @@ func (e *Edge) backhaul(w http.ResponseWriter, r *http.Request, host string) boo
e.logThrottled("backhaul-no-mesh-key", fmt.Sprintf("%s is homed on %s but no mesh key is mirrored — not dialling (the request gets the unknown-host answer until the key lands)", host, box))
return false
}
+ // Containment (the egress ruling, 2026-08-24), between the last guard and
+ // the dial: a saturated home box is answered here, with no dial and no
+ // bytes crossing a region. 503 + Retry-After: 2 is this package's
+ // established "come back shortly" (staleHopRefusal, fleetProxy) and is
+ // deliberately NOT the unknown-host answer — that is a 302 on a fallback
+ // edge, which would lie to a legitimate client about where its app lives.
+ // The log key is per home BOX, like "backhaul-error "+box below and for
+ // the same bounded-key reason.
+ release, ok := e.backhaulAcquire(box)
+ if !ok {
+ e.logThrottled("backhaul-saturated "+box,
+ fmt.Sprintf("backhaul to %s (%s) is at its in-flight cap (%d) — answering 503 + Retry-After: 2", box, eip, backhaulMaxInflight))
+ w.Header().Set("Retry-After", "2")
+ http.Error(w, "temporarily busy", http.StatusServiceUnavailable)
+ return true
+ }
+ // Held until the hop's ServeHTTP returns, which for a stream or an
+ // upgraded connection is the life of the stream (backhaulAcquire).
+ defer release()
// The MAC is over the CONCRETE request host — never the wildcard map key
// that may have matched it — because the home edge verifies against its own
// literal request host (hopMAC canonicalizes both, mesh.go).
@@ -2106,6 +2197,11 @@ func RunWithActivatorReady(args []string, act Activator, cache *Cache, ready fun
log.Fatalf("edge: tunnels: %v", err)
}
+ // The backhaul in-flight cap, read once here rather than per request
+ // (backhaulMaxInflight). Inert on a single-box deployment, where nothing
+ // is ever backhauled at all.
+ backhaulMaxInflight = backhaulInflightFromEnv()
+
go e.Watch(make(chan struct{})) // process-lifetime; never stopped
// Reload routes on SIGHUP too (an operator's explicit "now").
diff --git a/internal/edge/homemap_test.go b/internal/edge/homemap_test.go
index 8854054b2..74d00db80 100644
--- a/internal/edge/homemap_test.go
+++ b/internal/edge/homemap_test.go
@@ -1138,6 +1138,302 @@ func TestHTTPFallbackRedirectsARemoteHomedHost(t *testing.T) {
}
}
+// ---- containment: the per-home-box in-flight cap --------------------------
+
+// backhaulCap lowers the in-flight cap for one test and restores it after —
+// the package-seam pattern backhaulPort and homeMapInterval already follow.
+func backhaulCap(t *testing.T, n int) {
+ t.Helper()
+ old := backhaulMaxInflight
+ backhaulMaxInflight = n
+ t.Cleanup(func() { backhaulMaxInflight = old })
+}
+
+// backhaulGetter drives requests for host at one edge over a single reused
+// server, so several can be in flight at once (hopRequest stands a server per
+// call, which is fine for one-shot tests and useless for a concurrency one).
+func backhaulGetter(t *testing.T, e *Edge) func(host string) (*http.Response, error) {
+ t.Helper()
+ srv := httptest.NewServer(e)
+ t.Cleanup(srv.Close)
+ return func(host string) (*http.Response, error) {
+ req, err := http.NewRequest(http.MethodGet, srv.URL+"/", nil)
+ if err != nil {
+ return nil, err
+ }
+ req.Host = host
+ return http.DefaultClient.Do(req)
+ }
+}
+
+// The cap itself: with two hops held open toward one home box, a third is
+// answered 503 + Retry-After: 2 HERE — no dial, no cross-region bytes — and
+// the slot a finished hop gives back is immediately usable again.
+//
+// This is what bounds a scanner's forged-Host flood: egress is (concurrent
+// streams × per-stream bandwidth), and this caps the first factor per box.
+func TestBackhaulRefusesAtInflightCap(t *testing.T) {
+ e, m, _ := hopEdge(t, "annie", "local.test")
+ backhaulCap(t, 2)
+
+ started, hold := make(chan struct{}, 4), make(chan struct{})
+ rec, eip := fakeHomeEdge(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ started <- struct{}{}
+ <-hold
+ fmt.Fprint(w, homeEdgeBody)
+ }))
+ homedOn(t, m, "betty", eip, "example.com")
+ get := backhaulGetter(t, e)
+
+ done := make(chan error, 2)
+ for i := 0; i < 2; i++ {
+ go func() {
+ resp, err := get("example.com")
+ if err != nil {
+ done <- err
+ return
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusOK {
+ done <- fmt.Errorf("held hop: status = %d, want 200", resp.StatusCode)
+ return
+ }
+ body, err := io.ReadAll(resp.Body)
+ if err != nil {
+ done <- err
+ return
+ }
+ if string(body) != homeEdgeBody {
+ done <- fmt.Errorf("held hop: body = %q", body)
+ return
+ }
+ done <- nil
+ }()
+ }
+ for i := 0; i < 2; i++ {
+ select {
+ case <-started:
+ case err := <-done:
+ t.Fatalf("a hop finished before the cap was full: %v", err)
+ case <-time.After(15 * time.Second):
+ t.Fatal("the two hops that fill the cap never reached the home edge")
+ }
+ }
+
+ // Both slots are held. The third request must be refused by THIS box.
+ dialled := rec.get().n
+ resp, err := get("example.com")
+ if err != nil {
+ t.Fatalf("third request: %v", err)
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d, want 503 at the in-flight cap", resp.StatusCode)
+ }
+ if got := resp.Header.Get("Retry-After"); got != "2" {
+ t.Fatalf("Retry-After = %q, want 2 — the cap clears as hops finish", got)
+ }
+ if got := rec.get().n; got != dialled {
+ t.Fatalf("the home edge saw %d request(s), want %d — a refusal must cost no dial and no cross-region bytes", got, dialled)
+ }
+
+ // Released slots are usable again: the cap is a bound on CONCURRENCY, not
+ // a quota that a box burns through and never gets back.
+ close(hold)
+ for i := 0; i < 2; i++ {
+ select {
+ case err := <-done:
+ if err != nil {
+ t.Fatal(err)
+ }
+ case <-time.After(15 * time.Second):
+ t.Fatal("a held hop never completed after its home edge was released")
+ }
+ }
+ after, err := get("example.com")
+ if err != nil {
+ t.Fatalf("fourth request: %v", err)
+ }
+ defer after.Body.Close()
+ if after.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d after the held hops finished, want 200", after.StatusCode)
+ }
+ if got := rec.get().n; got != dialled+1 {
+ t.Fatalf("the home edge saw %d request(s), want %d — the fourth request must dial normally", got, dialled+1)
+ }
+}
+
+// The cap is per HOME BOX: one saturated box must not stop this edge serving
+// hosts homed on any other. A deployment-wide counter would let one busy (or
+// one attacked) box take the whole mesh down.
+func TestBackhaulInflightCapIsPerHomeBox(t *testing.T) {
+ e, m, _ := hopEdge(t, "annie", "local.test")
+ backhaulCap(t, 1)
+ rec, eip := fakeHomeEdge(t, nil)
+ homedOn(t, m, "carol", eip, "example.com")
+ // Betty's EIP is never dialled in this test — her slot is filled directly
+ // below, so the refusal happens before any dial. (TEST-NET-1: a dial
+ // would hang for the whole 5s budget and be visible as a failure.)
+ homedOn(t, m, "betty", "192.0.2.5", "saturated.test")
+
+ release, ok := e.backhaulAcquire("betty")
+ if !ok {
+ t.Fatal("the first slot toward a box must always be available")
+ }
+ t.Cleanup(release)
+
+ if resp := hopRequest(t, e, "saturated.test", nil); resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d, want 503 — betty's only slot is held", resp.StatusCode)
+ }
+ resp := hopRequest(t, e, "example.com", nil)
+ if resp.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d, want 200 — carol's cap is her own", resp.StatusCode)
+ }
+ if body := readBody(t, resp); body != homeEdgeBody {
+ t.Fatalf("body = %q, want the home edge's own answer", body)
+ }
+ if n := rec.get().n; n != 1 {
+ t.Fatalf("carol saw %d request(s), want exactly one", n)
+ }
+}
+
+// A slot is held for the LIFE of the stream, not until the response headers
+// arrive: a long-lived stream (a WebSocket, an SSE feed, a large download) is
+// precisely the sustained cross-region egress the cap exists to bound, so it
+// must count against the cap for as long as it flows.
+func TestBackhaulSlotHeldForStreamLife(t *testing.T) {
+ e, m, _ := hopEdge(t, "annie", "local.test")
+ backhaulCap(t, 1)
+
+ streaming, finish := make(chan struct{}), make(chan struct{})
+ _, eip := fakeHomeEdge(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ fmt.Fprint(w, "first ")
+ w.(http.Flusher).Flush() // headers and a first chunk are on the wire
+ close(streaming)
+ <-finish
+ fmt.Fprint(w, "last")
+ }))
+ homedOn(t, m, "betty", eip, "example.com")
+ get := backhaulGetter(t, e)
+
+ stream, err := get("example.com")
+ if err != nil {
+ t.Fatalf("stream request: %v", err)
+ }
+ defer stream.Body.Close()
+ if stream.StatusCode != http.StatusOK {
+ t.Fatalf("stream status = %d, want 200", stream.StatusCode)
+ }
+ select {
+ case <-streaming:
+ case <-time.After(15 * time.Second):
+ t.Fatal("the streaming hop never reached the home edge")
+ }
+
+ // Headers have arrived and the body is still flowing: the slot is held.
+ busy, err := get("example.com")
+ if err != nil {
+ t.Fatalf("second request: %v", err)
+ }
+ defer busy.Body.Close()
+ if busy.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d while a hop is mid-stream, want 503 — the slot is held for the stream's life", busy.StatusCode)
+ }
+
+ close(finish)
+ body, err := io.ReadAll(stream.Body)
+ if err != nil {
+ t.Fatalf("read stream: %v", err)
+ }
+ if string(body) != "first last" {
+ t.Fatalf("stream body = %q, want both chunks", body)
+ }
+ stream.Body.Close()
+
+ // And the moment it ends, the slot is back.
+ after, err := get("example.com")
+ if err != nil {
+ t.Fatalf("third request: %v", err)
+ }
+ defer after.Body.Close()
+ if after.StatusCode != http.StatusOK {
+ t.Fatalf("status = %d after the stream ended, want 200 — the slot must have been released", after.StatusCode)
+ }
+}
+
+// Saturation is one throttled journal line per home box, never silence and
+// never one line per refused request: a flood is a STANDING condition, and
+// the key is the box (bounded by the deployment) and never the host (which a
+// stranger picks, and which a catch-all home would multiply without limit).
+func TestBackhaulSaturationLogsThrottledPerBox(t *testing.T) {
+ e, m, _ := hopEdge(t, "annie", "local.test")
+ backhaulCap(t, 1)
+ // Neither EIP is ever dialled: both boxes' only slot is held below, so
+ // every request in this test is refused before the dial.
+ homedOn(t, m, "betty", "192.0.2.5", "betty.test")
+ homedOn(t, m, "carol", "192.0.2.6", "carol.test")
+ for _, box := range []string{"betty", "carol"} {
+ release, ok := e.backhaulAcquire(box)
+ if !ok {
+ t.Fatalf("%s: the first slot toward a box must always be available", box)
+ }
+ t.Cleanup(release)
+ }
+
+ buf := captureLog(t)
+ for _, host := range []string{"betty.test", "betty.test", "carol.test"} {
+ if resp := hopRequest(t, e, host, nil); resp.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("%s: status = %d, want 503", host, resp.StatusCode)
+ }
+ }
+
+ got := buf.String()
+ // logThrottled prints no key of its own, so key isolation is observed as
+ // TWO first occurrences (one bucket per box) with betty's second refusal
+ // throttled inside her own bucket.
+ if n := strings.Count(got, "occurrence #1"); n != 2 {
+ t.Fatalf("%d first-occurrence line(s), want one per home box (two buckets): %s", n, got)
+ }
+ if strings.Contains(got, "occurrence #2") {
+ t.Fatalf("log output = %q, want the second refusal against the same box throttled", got)
+ }
+ if n := strings.Count(got, "in-flight cap"); n != 2 {
+ t.Fatalf("%d saturation line(s), want exactly two: %s", n, got)
+ }
+ if !strings.Contains(got, "betty") || !strings.Contains(got, "carol") {
+ t.Fatalf("log output = %q, want each saturated home box named", got)
+ }
+}
+
+// The knob: unset, unparseable and non-positive all mean the default. There
+// is deliberately no "off" — an unbounded cross-region path is what the cap
+// closes — and a typo must not silently zero-cap a deployment either.
+func TestBackhaulMaxInflightFromEnv(t *testing.T) {
+ for _, tc := range []struct {
+ set bool
+ raw string
+ want int
+ }{
+ {set: false, want: defaultBackhaulMaxInflight},
+ {set: true, raw: "", want: defaultBackhaulMaxInflight},
+ {set: true, raw: " ", want: defaultBackhaulMaxInflight},
+ {set: true, raw: "sixty-four", want: defaultBackhaulMaxInflight},
+ {set: true, raw: "0", want: defaultBackhaulMaxInflight},
+ {set: true, raw: "-5", want: defaultBackhaulMaxInflight},
+ {set: true, raw: "128", want: 128},
+ {set: true, raw: " 8 ", want: 8},
+ } {
+ if tc.set {
+ t.Setenv("CARLOS_BACKHAUL_MAX_INFLIGHT", tc.raw)
+ } else {
+ os.Unsetenv("CARLOS_BACKHAUL_MAX_INFLIGHT")
+ }
+ if got := backhaulInflightFromEnv(); got != tc.want {
+ t.Fatalf("CARLOS_BACKHAUL_MAX_INFLIGHT=%q (set=%v) → %d, want %d", tc.raw, tc.set, got, tc.want)
+ }
+ }
+}
+
// oneKeyFailingStore fails Get for one named key — a single unreadable object
// mid-poll, the case the carry-forward exists for.
type oneKeyFailingStore struct {