carlos / platform Public

edge: probe mode refuses to forward and stamps the serving box on the final response through the real writer stack

A request with X-Carlos-Probe: local is refused in backhaulTo with 503
"not served here" before the mesh key and the in-flight cap, so no
forwarding path dials it (ordinary, an exact name another box wins over
this box's wildcard, a door call). A probe served locally runs under
servedByWriter (servedby.go), installed on the connection's writer
beneath the counting writer and the gzip writer: the final status
removes every X-Carlos-Served-By, sets this box's id and no-store; a 1xx
passes through uncommitted; Flush commits a 200 first; Hijack delegates
straight down; Unwrap serves http.ResponseController. A probe is kept out
of the response cache in both directions.

TestProbeModeNeverPassesABackhauledAnswer, TestProbeModeThroughTheRealStack,
TestServedByWriterContract, and TestFixtureHoldIsACandidateNeverServed
(owed from Task 4: the writer fixture's hold is a holding candidate,
never served).
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev 8582e1c3b45e095a2984f16698b982a2c59a6bf1 parent e9c26e2
3 files changed, +637 −1
  • internal/edge/edge.go +27 −1
  • internal/edge/probe_test.go +504 −0
  • internal/edge/servedby.go +106 −0
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index 8f4ac08ea..c56fb7054 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -1678,6 +1678,16 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
e.unknownHost(w, r)
return
}
+ // Probe mode (servedby.go): a probe this box serves names this box on
+ // its final response. The writer goes on first, under the counting
+ // writer and the gzip writer, so every local answer below (the gate's,
+ // the activator's, the app's) is stamped on the block the connection
+ // writes. The forwarding paths above never get here: backhaulTo
+ // refuses a probe before it dials.
+ probe := isProbe(r)
+ if probe {
+ w = &servedByWriter{ResponseWriter: w, box: e.boxID}
+ }
up := *selectUpstream(local, r.URL.Path)
// A login POST re-resolves its upstream by the path in `next`, not by
@@ -1881,7 +1891,12 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// A gated row's verify request is exempt in both directions too: it must
// reach the import rather than a response cached for the legacy route,
// and the import's answer must not outlive a rollback.
- cacheable := e.cache != nil && (r.Method == http.MethodGet || r.Method == http.MethodHead) &&
+ //
+ // A probe is exempt in both directions as well: it must reach the app
+ // (a cached answer says nothing about whether this box serves it now),
+ // and its stamped, no-store answer must not replace an entry ordinary
+ // visitors are served.
+ cacheable := !probe && e.cache != nil && (r.Method == http.MethodGet || r.Method == http.MethodHead) &&
up.route.Kind != "status" && up.route.ImportGate == ""
var key CacheKey
if cacheable {
@@ -2993,6 +3008,17 @@ func (e *Edge) backhaulTo(w http.ResponseWriter, r *http.Request, host, eip, box
m.staleHopRefusal(w, host)
return true
}
+ // Probe mode (servedby.go): a probe is answered by the box it reached
+ // or refused, never forwarded, so a backhauled success cannot pass the
+ // console's health check. Every forwarding path comes through here (an
+ // ordinary request, an exact name another box wins over this box's
+ // wildcard, a door call), and the refusal comes before the key, so a
+ // missing key does not turn it into the local fallback, and before the
+ // cap, so it spends no slot.
+ if isProbe(r) {
+ probeRefusal(w)
+ return true
+ }
// Sign with the mirror's cached pair, .next preferred (signingKey, mesh.go)
// — never a store read on the request path. NO KEY MEANS NO DIAL: signHop
// would happily HMAC under an empty key, the home edge would refuse the
diff --git a/internal/edge/probe_test.go b/internal/edge/probe_test.go
new file mode 100644
index 000000000..53c0e525b
--- /dev/null
+++ b/internal/edge/probe_test.go
@@ -0,0 +1,504 @@
+package edge
+
+// Probe mode (self-serve moves plan 3 Task 6): a request carrying
+// X-Carlos-Probe: local is never forwarded, and when this box serves it
+// the final response names this box in X-Carlos-Served-By, through the
+// writer stack every ordinary request goes through, with the cache
+// bypassed in both directions.
+
+import (
+ "bufio"
+ "compress/gzip"
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "net"
+ "net/http"
+ "net/http/httptest"
+ "net/http/httptrace"
+ "net/textproto"
+ "os"
+ "slices"
+ "strconv"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// probeBox is this box's id in these tests, the shape plan 2 names boxes by.
+const probeBox = "i-0aaa"
+
+// serveProbe is sc.serve with the probe header set when probe is true.
+func (sc *scene) serveProbe(host string, v via, probe bool) *httptest.ResponseRecorder {
+ sc.t.Helper()
+ r := httptest.NewRequest(http.MethodGet, "https://"+host+"/", nil)
+ r.Host = host
+ if probe {
+ r.Header.Set(probeHeader, probeLocal)
+ }
+ if v.door != nil {
+ r = r.WithContext(context.WithValue(r.Context(), doorCallerKey{}, *v.door))
+ }
+ w := httptest.NewRecorder()
+ sc.e.ServeHTTP(w, r)
+ return w
+}
+
+func wantProbeRefusal(t *testing.T, w *httptest.ResponseRecorder, what string) {
+ t.Helper()
+ if w.Code != http.StatusServiceUnavailable || strings.TrimSpace(w.Body.String()) != "not served here" {
+ t.Fatalf("%s = %d %q, want 503 \"not served here\"", what, w.Code, w.Body.String())
+ }
+ if got := w.Header().Values(servedByHeader); len(got) != 0 {
+ t.Fatalf("%s: a refusal names a serving box %q", what, got)
+ }
+}
+
+// wantHop checks the home edge's last request was a signed hop for host
+// whose forwarding chain names only the client.
+func wantHop(t *testing.T, sc *scene, rec *homeEdgeRecord, host, what string) {
+ t.Helper()
+ seen := rec.get()
+ if seen.hop != "1" || !verifyHopEitherKey(sc.key, nil, host, seen.auth, time.Now()) {
+ t.Fatalf("%s: hop %q auth %q, want a hop signed for %s", what, seen.hop, seen.auth, host)
+ }
+ if seen.host != host || len(seen.xff) != 1 || strings.Contains(seen.xff[0], ",") {
+ t.Fatalf("%s: Host %q X-Forwarded-For %q, want %s and the client alone", what, seen.host, seen.xff, host)
+ }
+}
+
+// TestProbeModeNeverPassesABackhauledAnswer: on each of the three paths
+// that forward (an ordinary request for another box's host, an exact name
+// another box wins over this box's wildcard, a door call for its own app's
+// host on another box), a probe is answered 503 "not served here" and
+// nothing is dialled; without the header the same requests still forward
+// as before: a signed hop, the client's forwarding chain, the no-key
+// fallback and the in-flight cap.
+func TestProbeModeNeverPassesABackhauledAnswer(t *testing.T) {
+ t.Run("ordinary", func(t *testing.T) {
+ sc := newScene(t, "annie")
+ rec, eip := fakeHomeEdge(t, nil)
+ sc.publish("betty", eip, readyJSON("shop.example.com", "bab", "shop", 1, 0), "")
+ wantProbeRefusal(t, sc.serveProbe("shop.example.com", ordinary, true), "a probe")
+ if n := rec.get().n; n != 0 {
+ t.Fatalf("a probe dialled the home edge %d time(s)", n)
+ }
+ wantBody(t, sc.serveProbe("shop.example.com", ordinary, false), homeEdgeBody, "without the header")
+ wantHop(t, sc, rec, "shop.example.com", "without the header")
+
+ // Containment: at the cap an ordinary request is the busy 503,
+ // and a probe is still refused before the cap is consulted.
+ backhaulCap(t, 0)
+ wantProbeRefusal(t, sc.serveProbe("shop.example.com", ordinary, true), "a probe at the cap")
+ w := sc.serveProbe("shop.example.com", ordinary, false)
+ if w.Code != http.StatusServiceUnavailable || w.Header().Get("Retry-After") != "2" || !strings.Contains(w.Body.String(), "temporarily busy") {
+ t.Fatalf("at the cap = %d %q, want the containment 503", w.Code, w.Body.String())
+ }
+ if n := rec.get().n; n != 1 {
+ t.Fatalf("home edge saw %d, want 1", n)
+ }
+ })
+ t.Run("wildcard-exact", func(t *testing.T) {
+ sc := newScene(t, "annie")
+ rec, eip := fakeHomeEdge(t, nil)
+ sc.route("*.example.com", "local wildcard", "bab", "wild")
+ sc.publish("betty", eip, readyJSON("x.example.com", "bab", "shop", 1, 0), "")
+ if d := sc.e.resolve("x.example.com", sc.now); d.kind != decRemote || d.fallback == nil {
+ t.Fatalf("fixture: decision %v fallback %v, want decRemote over the local wildcard", d.kind, d.fallback)
+ }
+ wantProbeRefusal(t, sc.serveProbe("x.example.com", ordinary, true), "a probe")
+ if n := rec.get().n; n != 0 {
+ t.Fatalf("a probe dialled the home edge %d time(s)", n)
+ }
+ wantBody(t, sc.serveProbe("x.example.com", ordinary, false), homeEdgeBody, "without the header")
+ wantHop(t, sc, rec, "x.example.com", "without the header")
+
+ // No mesh key: an ordinary request falls back to the local
+ // wildcard; a probe is still refused, since the decision names
+ // another box as the winner.
+ sc.dropKey()
+ wantBody(t, sc.serveProbe("x.example.com", ordinary, false), "local wildcard", "without the header and no key")
+ wantProbeRefusal(t, sc.serveProbe("x.example.com", ordinary, true), "a probe with no key")
+ if n := rec.get().n; n != 1 {
+ t.Fatalf("home edge saw %d, want 1", n)
+ }
+ })
+ t.Run("door-authorised", func(t *testing.T) {
+ sc := newScene(t, "annie")
+ rec, eip := fakeHomeEdge(t, nil)
+ sc.publish("betty", eip, readyJSON("shop.example.com", "bab", "shop", 1, 0), "")
+ bab := via{door: &doorCaller{Account: "bab", App: "shop"}}
+ wantProbeRefusal(t, sc.serveProbe("shop.example.com", bab, true), "a door's probe")
+ if n := rec.get().n; n != 0 {
+ t.Fatalf("a door's probe dialled the home edge %d time(s)", n)
+ }
+ wantBody(t, sc.serveProbe("shop.example.com", bab, false), homeEdgeBody, "a door without the header")
+ wantHop(t, sc, rec, "shop.example.com", "a door without the header")
+ })
+}
+
+// probeStack is one edge with gzip and the response cache on, serving
+// shop.example.com from a backend that counts what reaches it, through a
+// real HTTP server.
+type probeStack struct {
+ sc *scene
+ url string
+ mu sync.Mutex
+ hits map[string]int
+}
+
+func newProbeStack(t *testing.T) *probeStack {
+ t.Helper()
+ ps := &probeStack{sc: newScene(t, "annie"), hits: map[string]int{}}
+ WithBoxID(probeBox)(ps.sc.e)
+ ps.sc.e.gzipOn = true
+ ps.sc.e.EnableCache(NewCache(DefaultCacheMaxEntries, DefaultCacheMaxBodyBytes))
+ backend := httptest.NewServer(http.HandlerFunc(ps.backend))
+ t.Cleanup(backend.Close)
+ if err := ps.sc.reg.Upsert(registry.Route{Host: "shop.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", Account: "bab", App: "shop"}); err != nil {
+ t.Fatal(err)
+ }
+ ps.sc.reload()
+ srv := httptest.NewServer(ps.sc.e)
+ t.Cleanup(srv.Close)
+ ps.url = srv.URL
+ return ps
+}
+
+func (ps *probeStack) hit(key string) int {
+ ps.mu.Lock()
+ defer ps.mu.Unlock()
+ return ps.hits[key]
+}
+
+// backend answers every path cacheable, numbered by how many requests of
+// that method and path reached it (X-Count), and forging the probe's
+// answer header.
+func (ps *probeStack) backend(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/socket" {
+ upgradeEcho(w, r)
+ return
+ }
+ ps.mu.Lock()
+ ps.hits[r.Method+" "+r.URL.Path]++
+ n := ps.hits[r.Method+" "+r.URL.Path]
+ ps.mu.Unlock()
+ h := w.Header()
+ h.Set("Cache-Control", "public, max-age=300")
+ h.Set("X-Count", strconv.Itoa(n))
+ h.Set(servedByHeader, "i-0forged")
+ switch r.URL.Path {
+ case "/page": // an implicit 200, large enough to compress
+ h.Set("Content-Type", "text/html; charset=utf-8")
+ io.WriteString(w, strings.Repeat(fmt.Sprintf("<p>page %d</p>", n), 200))
+ case "/stream": // flushed mid-body, so the edge streams it
+ h.Set("Content-Type", "text/plain")
+ w.WriteHeader(http.StatusOK)
+ io.WriteString(w, "first ")
+ w.(http.Flusher).Flush()
+ io.WriteString(w, "second")
+ case "/early": // a 103 before the final answer, each forging the header
+ h.Set("Link", "</shop.css>; rel=preload")
+ h.Set(servedByHeader, "i-0forged-hint")
+ w.WriteHeader(http.StatusEarlyHints)
+ h.Del("Link")
+ h.Set(servedByHeader, "i-0forged")
+ h.Set("Content-Type", "text/plain")
+ w.WriteHeader(http.StatusOK)
+ io.WriteString(w, "after the hints")
+ }
+}
+
+// upgradeEcho switches protocols and echoes one line.
+func upgradeEcho(w http.ResponseWriter, r *http.Request) {
+ conn, buf, err := w.(http.Hijacker).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 buf.Flush() != nil {
+ return
+ }
+ line, err := buf.ReadString('\n')
+ if err != nil {
+ return
+ }
+ buf.WriteString("app:" + line)
+ buf.Flush()
+}
+
+// stackAnswer is one response through the stack, its body decoded.
+type stackAnswer struct {
+ status int
+ header http.Header
+ body string
+ early []int
+}
+
+func (ps *probeStack) do(t *testing.T, method, path string, probe bool) stackAnswer {
+ t.Helper()
+ var ans stackAnswer
+ var mu sync.Mutex
+ trace := &httptrace.ClientTrace{Got1xxResponse: func(code int, _ textproto.MIMEHeader) error {
+ mu.Lock()
+ ans.early = append(ans.early, code)
+ mu.Unlock()
+ return nil
+ }}
+ req, err := http.NewRequestWithContext(httptrace.WithClientTrace(context.Background(), trace), method, ps.url+path, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Host = "shop.example.com"
+ req.Header.Set("Accept-Encoding", "gzip")
+ if probe {
+ req.Header.Set(probeHeader, probeLocal)
+ }
+ client := &http.Client{Transport: &http.Transport{DisableCompression: true}}
+ resp, err := client.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer resp.Body.Close()
+ var rd io.Reader = resp.Body
+ if resp.Header.Get("Content-Encoding") == "gzip" {
+ zr, err := gzip.NewReader(resp.Body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ rd = zr
+ }
+ b, err := io.ReadAll(rd)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ans.status, ans.header, ans.body = resp.StatusCode, resp.Header, string(b)
+ return ans
+}
+
+// TestProbeModeThroughTheRealStack: with gzip and the cache on and an entry
+// already cached for each URL, a probe reaches the backend, and its final
+// response names this box (one value, the app's forged one gone, no-store),
+// whether the 200 is implicit, a HEAD, flushed mid-body, or follows a 103
+// that forged the header too; the probe's answer is not stored, so a later
+// ordinary request still gets the old entry. An upgrade still hijacks,
+// with and without the header.
+func TestProbeModeThroughTheRealStack(t *testing.T) {
+ ps := newProbeStack(t)
+ cases := []struct {
+ method, path, body string
+ gzipped bool
+ early bool
+ }{
+ {http.MethodGet, "/page", strings.Repeat("<p>page 2</p>", 200), true, false},
+ {http.MethodHead, "/page", "", false, false},
+ {http.MethodGet, "/stream", "first second", true, false},
+ {http.MethodGet, "/early", "after the hints", false, true},
+ }
+ for _, c := range cases {
+ what := c.method + " " + c.path
+ // The cache entry the probe must not be answered from.
+ first := ps.do(t, c.method, c.path, false)
+ again := ps.do(t, c.method, c.path, false)
+ if first.status != 200 || again.header.Get("X-Count") != "1" || ps.hit(what) != 1 {
+ t.Fatalf("%s: fixture: %d, count %q, backend %d; want the second request answered from the cache", what, first.status, again.header.Get("X-Count"), ps.hit(what))
+ }
+
+ got := ps.do(t, c.method, c.path, true)
+ if got.status != http.StatusOK || ps.hit(what) != 2 || got.header.Get("X-Count") != "2" {
+ t.Fatalf("%s: probe = %d count %q, backend %d; want the backend's answer, not the cache's", what, got.status, got.header.Get("X-Count"), ps.hit(what))
+ }
+ if v := got.header.Values(servedByHeader); !slices.Equal(v, []string{probeBox}) {
+ t.Fatalf("%s: %s = %q, want exactly [%s]", what, servedByHeader, v, probeBox)
+ }
+ if cc := got.header.Values("Cache-Control"); !slices.Equal(cc, []string{"no-store"}) {
+ t.Fatalf("%s: Cache-Control = %q, want no-store", what, cc)
+ }
+ if got.body != c.body {
+ t.Fatalf("%s: body %q, want %q", what, got.body, c.body)
+ }
+ if (got.header.Get("Content-Encoding") == "gzip") != c.gzipped {
+ t.Fatalf("%s: Content-Encoding %q, gzip wanted %v", what, got.header.Get("Content-Encoding"), c.gzipped)
+ }
+ if c.early && !slices.Equal(got.early, []int{http.StatusEarlyHints}) {
+ t.Fatalf("%s: informational responses %v, want one 103", what, got.early)
+ }
+
+ later := ps.do(t, c.method, c.path, false)
+ if later.header.Get("X-Count") != "1" || ps.hit(what) != 2 {
+ t.Fatalf("%s: after the probe: count %q, backend %d; want the old entry", what, later.header.Get("X-Count"), ps.hit(what))
+ }
+ if v := later.header.Get(servedByHeader); v == probeBox {
+ t.Fatalf("%s: the cached answer names this box: the probe's response was stored", what)
+ }
+ }
+
+ for _, probe := range []bool{false, true} {
+ c, err := net.Dial("tcp", strings.TrimPrefix(ps.url, "http://"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ c.SetDeadline(time.Now().Add(10 * time.Second))
+ hdr := ""
+ if probe {
+ hdr = probeHeader + ": " + probeLocal + "\r\n"
+ }
+ fmt.Fprintf(c, "GET /socket HTTP/1.1\r\nHost: shop.example.com\r\nConnection: Upgrade\r\nUpgrade: websocket\r\n%s\r\n", hdr)
+ br := bufio.NewReader(c)
+ resp, err := http.ReadResponse(br, nil)
+ if err != nil || resp.StatusCode != http.StatusSwitchingProtocols {
+ c.Close()
+ t.Fatalf("upgrade (probe %v): %v %v, want 101", probe, resp, err)
+ }
+ fmt.Fprint(c, "ping\n")
+ line, err := br.ReadString('\n')
+ c.Close()
+ if err != nil || line != "app:ping\n" {
+ t.Fatalf("upgrade (probe %v): echo %q %v", probe, line, err)
+ }
+ }
+}
+
+// fakeWriter records what reaches the bottom of the stack.
+type fakeWriter struct {
+ h http.Header
+ codes []int
+ headers []http.Header
+ body strings.Builder
+ flushes int
+ hijacked bool
+ duplex bool
+}
+
+func newFakeWriter() *fakeWriter { return &fakeWriter{h: http.Header{}} }
+
+func (f *fakeWriter) Header() http.Header { return f.h }
+func (f *fakeWriter) WriteHeader(code int) {
+ f.codes = append(f.codes, code)
+ f.headers = append(f.headers, f.h.Clone())
+}
+func (f *fakeWriter) Write(p []byte) (int, error) {
+ if len(f.codes) == 0 {
+ f.WriteHeader(http.StatusOK)
+ }
+ return f.body.Write(p)
+}
+func (f *fakeWriter) Flush() { f.flushes++ }
+
+var errFakeHijack = errors.New("the fake's hijack")
+
+func (f *fakeWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+ f.hijacked = true
+ return nil, nil, errFakeHijack
+}
+func (f *fakeWriter) EnableFullDuplex() error { f.duplex = true; return nil }
+
+// stamped reports whether the i-th header block named only box and no-store.
+func (f *fakeWriter) stamped(i int, box string) bool {
+ return i < len(f.headers) && slices.Equal(f.headers[i].Values(servedByHeader), []string{box}) &&
+ slices.Equal(f.headers[i].Values("Cache-Control"), []string{"no-store"})
+}
+
+// TestServedByWriterContract: the writer's own rules, one at a time.
+func TestServedByWriterContract(t *testing.T) {
+ t.Run("an implicit 200 is stamped", func(t *testing.T) {
+ f := newFakeWriter()
+ w := &servedByWriter{ResponseWriter: f, box: probeBox}
+ w.Header().Add(servedByHeader, "i-0forged")
+ w.Header().Add(servedByHeader, "i-0another")
+ io.WriteString(w, "x")
+ if !slices.Equal(f.codes, []int{200}) || !f.stamped(0, probeBox) || f.body.String() != "x" {
+ t.Fatalf("codes %v headers %v body %q", f.codes, f.headers, f.body.String())
+ }
+ })
+ t.Run("a 1xx passes through and the final status is stamped", func(t *testing.T) {
+ f := newFakeWriter()
+ w := &servedByWriter{ResponseWriter: f, box: probeBox}
+ w.Header().Set("Link", "</a.css>; rel=preload")
+ w.Header().Set(servedByHeader, "i-0forged-hint")
+ w.WriteHeader(http.StatusEarlyHints)
+ if w.final || len(f.codes) != 1 || f.headers[0].Get("Link") == "" || f.headers[0].Get(servedByHeader) != "" {
+ t.Fatalf("after the 103: final %v codes %v headers %v", w.final, f.codes, f.headers)
+ }
+ w.Header().Set(servedByHeader, "i-0forged")
+ w.WriteHeader(http.StatusNotFound)
+ w.Header().Set(servedByHeader, "i-0late")
+ w.WriteHeader(http.StatusOK) // superfluous
+ if !slices.Equal(f.codes, []int{103, 404, 200}) || !f.stamped(1, probeBox) {
+ t.Fatalf("codes %v headers %v", f.codes, f.headers)
+ }
+ })
+ t.Run("a flush before any header commits a stamped 200", func(t *testing.T) {
+ f := newFakeWriter()
+ w := &servedByWriter{ResponseWriter: f, box: probeBox}
+ w.Header().Set(servedByHeader, "i-0forged")
+ w.Flush()
+ if !slices.Equal(f.codes, []int{200}) || !f.stamped(0, probeBox) || f.flushes != 1 {
+ t.Fatalf("codes %v headers %v flushes %d", f.codes, f.headers, f.flushes)
+ }
+ w.Flush()
+ if len(f.codes) != 1 || f.flushes != 2 {
+ t.Fatalf("a second flush: codes %v flushes %d", f.codes, f.flushes)
+ }
+ })
+ t.Run("a flush over a writer that cannot flush still commits", func(t *testing.T) {
+ rec := httptest.NewRecorder()
+ w := &servedByWriter{ResponseWriter: struct{ http.ResponseWriter }{rec}, box: probeBox}
+ w.Flush()
+ if rec.Code != 200 || rec.Header().Get(servedByHeader) != probeBox {
+ t.Fatalf("code %d header %q", rec.Code, rec.Header().Get(servedByHeader))
+ }
+ })
+ t.Run("hijack delegates and commits nothing", func(t *testing.T) {
+ f := newFakeWriter()
+ w := &servedByWriter{ResponseWriter: f, box: probeBox}
+ cw := &countingWriter{ResponseWriter: w, n: new(int64)}
+ if _, _, err := cw.Hijack(); !errors.Is(err, errFakeHijack) || !f.hijacked || len(f.codes) != 0 {
+ t.Fatalf("err %v hijacked %v codes %v", err, f.hijacked, f.codes)
+ }
+ plain := &servedByWriter{ResponseWriter: struct{ http.ResponseWriter }{httptest.NewRecorder()}, box: probeBox}
+ if _, _, err := plain.Hijack(); err == nil {
+ t.Fatal("a writer that cannot hijack hijacked")
+ }
+ })
+ t.Run("Unwrap reaches the writer below", func(t *testing.T) {
+ f := newFakeWriter()
+ w := &servedByWriter{ResponseWriter: f, box: probeBox}
+ if w.Unwrap() != http.ResponseWriter(f) {
+ t.Fatal("Unwrap is not the writer below")
+ }
+ if err := http.NewResponseController(w).EnableFullDuplex(); err != nil || !f.duplex {
+ t.Fatalf("ResponseController: %v, duplex %v", err, f.duplex)
+ }
+ })
+}
+
+// TestFixtureHoldIsACandidateNeverServed (owed from plan 3 Task 4): a home
+// map refreshed from the writer's fixture offers its hold and the hold's
+// alias as holding hosts and serves only the file's route: the hold is a
+// candidate, never served evidence.
+func TestFixtureHoldIsACandidateNeverServed(t *testing.T) {
+ body, err := os.ReadFile("../edgesfile/testdata/offline-moves.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ s := meshTestStore(t)
+ putJSON(t, s, "control/edges/annie.json", string(body))
+ m := newHomeMap(s, "betty")
+ at := time.Unix(1792832400, 0)
+ m.now = func() time.Time { return at }
+ refreshHomeMap(t, m)
+ holding := m.HoldingHosts(at)
+ slices.Sort(holding)
+ if want := []string{"cart.bab.example.net", "cart.example.com"}; !slices.Equal(holding, want) {
+ t.Fatalf("HoldingHosts = %q, want %q", holding, want)
+ }
+ if served, want := m.ServedHosts(), []string{"shop.example.com"}; !slices.Equal(served, want) {
+ t.Fatalf("ServedHosts = %q, want %q", served, want)
+ }
+}
diff --git a/internal/edge/servedby.go b/internal/edge/servedby.go
new file mode 100644
index 000000000..cd03a6916
--- /dev/null
+++ b/internal/edge/servedby.go
@@ -0,0 +1,106 @@
+package edge
+
+// Probe mode (self-serve moves plan 3 Task 6; spec §7.4). The console's
+// health probe asks the target box directly with X-Carlos-Probe: local. An
+// edge that would forward such a request answers probeRefusal instead, so a
+// backhauled success can never pass for the target's own. An edge that
+// serves it locally names itself on the final response through
+// servedByWriter, after removing any header of that name the app set.
+
+import (
+ "bufio"
+ "fmt"
+ "net"
+ "net/http"
+)
+
+const (
+ probeHeader = "X-Carlos-Probe"
+ probeLocal = "local"
+ servedByHeader = "X-Carlos-Served-By"
+)
+
+// isProbe reports whether r asks to be answered by this box or not at all.
+func isProbe(r *http.Request) bool { return r.Header.Get(probeHeader) == probeLocal }
+
+// probeRefusal is the answer to a probe this box would have forwarded. No
+// Retry-After: the console's probe has its own cadence, and nothing about
+// this box changes for a retry.
+func probeRefusal(w http.ResponseWriter) {
+ w.Header().Set("Cache-Control", "no-store")
+ http.Error(w, "not served here", http.StatusServiceUnavailable)
+}
+
+// servedByWriter stamps the box id on a probe's final response. ServeHTTP
+// installs it under every other wrapper (directly on the connection's
+// writer, before the counting writer and the gzip writer wrap it), so the
+// header block it edits is the one the connection writes: the gzip
+// writer's private map has been copied down by then, and nothing below
+// it can add the header back.
+//
+// An informational status (a 103) passes through without committing: the
+// final status is still to come, and it is the one stamped. A forged copy
+// of the header is removed from it too, so no block of the response names
+// a box this one did not vouch for.
+type servedByWriter struct {
+ http.ResponseWriter
+ box string
+ final bool
+}
+
+// commit stamps the header block once, for the final status. A box that
+// was given no id (WithBoxID unset) names none, and the app's copy is
+// still removed: the console's check then fails, as it should.
+func (s *servedByWriter) commit() {
+ s.final = true
+ h := s.ResponseWriter.Header()
+ h.Del(servedByHeader)
+ if s.box != "" {
+ h.Set(servedByHeader, s.box)
+ }
+ h.Set("Cache-Control", "no-store")
+}
+
+func (s *servedByWriter) WriteHeader(code int) {
+ if code < 200 {
+ s.ResponseWriter.Header().Del(servedByHeader)
+ s.ResponseWriter.WriteHeader(code)
+ return
+ }
+ if !s.final {
+ s.commit()
+ }
+ s.ResponseWriter.WriteHeader(code)
+}
+
+func (s *servedByWriter) Write(p []byte) (int, error) {
+ if !s.final {
+ s.WriteHeader(http.StatusOK)
+ }
+ return s.ResponseWriter.Write(p)
+}
+
+// Flush commits a 200 when nothing has, as net/http's own Flush does, so
+// the stamped block is the one a flush sends.
+func (s *servedByWriter) Flush() {
+ if !s.final {
+ s.WriteHeader(http.StatusOK)
+ }
+ if f, ok := s.ResponseWriter.(http.Flusher); ok {
+ f.Flush()
+ }
+}
+
+// Hijack hands over the connection without committing anything, straight
+// to the writer below, so the counting writer's assertion above finds a
+// hijacker exactly when the connection is one.
+func (s *servedByWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+ hj, ok := s.ResponseWriter.(http.Hijacker)
+ if !ok {
+ return nil, nil, fmt.Errorf("edge: underlying ResponseWriter does not support http.Hijacker")
+ }
+ return hj.Hijack()
+}
+
+// Unwrap is for http.ResponseController.
+func (s *servedByWriter) Unwrap() http.ResponseWriter { return s.ResponseWriter }