amadan

carlos / platform Public

edge: correct the cap's doc, and harden its hot path (review)

Review findings on the in-flight cap, all mechanical:

- backhaulAcquire's doc claimed a property the code does not have
  ("under-admission is not attempted... one of them refuses"). What it
  actually guarantees: the count is incremented FIRST and rolled back on
  refusal, so the cap can never be exceeded — two racers at the limit
  both refuse, and a request arriving inside another's rollback window
  can be refused while a slot is free. Both errors are in the safe
  direction; the doc now says so.
- The named return is releaseSlot, not release: the old name shadowed
  the imported release package inside two functions of a file that uses
  it elsewhere.
- Load before LoadOrStore, the backhaulTransport idiom, so the hit path
  does not allocate an atomic.Int64 per backhauled request just to throw
  it away.
- backhaulMaxInflight is an atomic.Int64. It is written once before any
  listener binds today, but it is READ by every backhauling goroutine —
  a future env re-read, or a second Edge composed in one process, would
  turn the plain int into a real data race rather than a stale read.
- Tests: the env test captures and restores an operator's ambient
  CARLOS_BACKHAUL_MAX_INFLIGHT instead of clearing it, and the
  released-slot assertion retries for a bounded window rather than
  racing the deferred release on its first attempt.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev e3d2feeb6f6f5ac23a6f43ab82d7b5cfce23501f parent f126ca0
2 files changed, +78 −31
  • internal/edge/edge.go +34 −19
  • internal/edge/homemap_test.go +44 −12
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index bfd0fda66..affdadd91 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -1823,10 +1823,17 @@ func fleetProxy(fleet, boxLabel string, hub *TunnelHub) *httputil.ReverseProxy {
// 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
+//
+// Atomic rather than a plain int because it is READ on the request path by
+// every backhauling goroutine: it is written once today, before any listener
+// binds, but a future env re-read (or a second Edge composed in one process)
+// would turn a plain int into a real data race rather than a stale read.
+var backhaulMaxInflight atomic.Int64
const defaultBackhaulMaxInflight = 64
+func init() { backhaulMaxInflight.Store(defaultBackhaulMaxInflight) }
+
// 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.
@@ -1844,20 +1851,28 @@ func backhaulInflightFromEnv() int {
}
// 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.
+// saturation. releaseSlot 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))
+// stranger controls (backhaulInflight's own doc). The count is incremented
+// FIRST and rolled back on refusal, so the cap can never be exceeded: two
+// requests racing at the limit both see the excess and both refuse, and a
+// request arriving inside another's rollback window can be refused while a
+// slot is in fact free. Both errors are in the safe direction — a refusal is
+// a 503 the client retries in two seconds, where an over-admission would be
+// cross-region bytes the cap exists to prevent.
+func (e *Edge) backhaulAcquire(box string) (releaseSlot func(), ok bool) {
+ // Load first, like backhaulTransport: LoadOrStore would allocate a
+ // counter per backhauled request just to throw it away on the hit path.
+ v, hit := e.backhaulInflight.Load(box)
+ if !hit {
+ v, _ = e.backhaulInflight.LoadOrStore(box, new(atomic.Int64))
+ }
n := v.(*atomic.Int64)
- if n.Add(1) > int64(backhaulMaxInflight) {
+ if n.Add(1) > backhaulMaxInflight.Load() {
n.Add(-1)
return nil, false
}
@@ -1943,17 +1958,17 @@ func (e *Edge) backhaul(w http.ResponseWriter, r *http.Request, host string) boo
// 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)
+ releaseSlot, 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))
+ fmt.Sprintf("backhaul to %s (%s) is at its in-flight cap (%d) — answering 503 + Retry-After: 2", box, eip, backhaulMaxInflight.Load()))
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()
+ defer releaseSlot()
// 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).
@@ -2197,10 +2212,10 @@ 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()
+ // The backhaul in-flight cap, read from the environment once here rather
+ // than per request (backhaulMaxInflight). Inert on a single-box
+ // deployment, where nothing is ever backhauled at all.
+ backhaulMaxInflight.Store(int64(backhaulInflightFromEnv()))
go e.Watch(make(chan struct{})) // process-lifetime; never stopped
diff --git a/internal/edge/homemap_test.go b/internal/edge/homemap_test.go
index 74d00db80..aa87a7df5 100644
--- a/internal/edge/homemap_test.go
+++ b/internal/edge/homemap_test.go
@@ -1142,11 +1142,11 @@ func TestHTTPFallbackRedirectsARemoteHomedHost(t *testing.T) {
// 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) {
+func backhaulCap(t *testing.T, n int64) {
t.Helper()
- old := backhaulMaxInflight
- backhaulMaxInflight = n
- t.Cleanup(func() { backhaulMaxInflight = old })
+ old := backhaulMaxInflight.Load()
+ backhaulMaxInflight.Store(n)
+ t.Cleanup(func() { backhaulMaxInflight.Store(old) })
}
// backhaulGetter drives requests for host at one edge over a single reused
@@ -1250,16 +1250,36 @@ func TestBackhaulRefusesAtInflightCap(t *testing.T) {
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)
+ // Retried for a bounded window rather than asserted on the first attempt:
+ // a held hop's client-side goroutine reports completion when the BODY is
+ // read, which can be a hair before the edge's own deferred release runs.
+ // The property under test is that the slots come back, not that they come
+ // back within one scheduling quantum.
+ var (
+ after *http.Response
+ lastErr error
+ )
+ deadline := time.Now().Add(10 * time.Second)
+ for {
+ after, lastErr = get("example.com")
+ if lastErr != nil {
+ t.Fatalf("fourth request: %v", lastErr)
+ }
+ if after.StatusCode == http.StatusOK {
+ break
+ }
+ after.Body.Close()
+ if after.StatusCode != http.StatusServiceUnavailable {
+ t.Fatalf("status = %d after the held hops finished, want 200", after.StatusCode)
+ }
+ if time.Now().After(deadline) {
+ t.Fatal("the cap never cleared after both held hops finished — a released slot was not given back")
+ }
+ time.Sleep(10 * time.Millisecond)
}
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)
+ if got := rec.get().n; got < dialled+1 {
+ t.Fatalf("the home edge saw %d request(s), want at least %d — the fourth request must dial normally", got, dialled+1)
}
}
@@ -1409,6 +1429,18 @@ func TestBackhaulSaturationLogsThrottledPerBox(t *testing.T) {
// 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) {
+ // The ambient value is captured and restored by hand: t.Setenv restores
+ // what IT set, but the unset case below has to clear the variable, and an
+ // operator's own CARLOS_BACKHAUL_MAX_INFLIGHT must survive this test.
+ ambient, had := os.LookupEnv("CARLOS_BACKHAUL_MAX_INFLIGHT")
+ t.Cleanup(func() {
+ if had {
+ os.Setenv("CARLOS_BACKHAUL_MAX_INFLIGHT", ambient)
+ return
+ }
+ os.Unsetenv("CARLOS_BACKHAUL_MAX_INFLIGHT")
+ })
+
for _, tc := range []struct {
set bool
raw string