amadan

carlos / platform Public

edge: the counting writer observes status; upstream 5xx become edge errors

Installed unconditionally at the one site every response shape passes;
the activator marks its own wake-path 503s so they are filed once, as the
platform log error they already are (spec 2026-09-07 §3.2–3.3).

The answer flag lives in internal/edge, not internal/activator as the
plan had it: the activator already imports the edge (backend_exec.go,
for edge.Cache and edge.ListenControl), so the flag there would have
been an import cycle. The activator calls edge.MarkAnswered; the stale
doc comment claiming the dependency points edge→activator is corrected
to match the code.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev f5ef07b21db0a33e396b6c20176f549bfe748a3f parent 2a3756f
8 files changed, +327 −20
  • internal/activator/activator.go +14 −2
  • internal/edge/answered.go +37 −0
  • internal/edge/answered_test.go +19 −0
  • internal/edge/edge.go +77 −13
  • internal/edge/edgeerrors_test.go +113 −0
  • internal/edge/usagemeter.go +34 −5
  • internal/edge/usagemeter_test.go +26 −0
  • internal/host/agent.go +7 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f1b545544..dd5fbbf24 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -54,6 +54,7 @@ import (
"sync"
"time"
+ "github.com/carlosframework/platform/internal/edge"
"github.com/carlosframework/platform/internal/registry"
)
@@ -371,8 +372,10 @@ func (e *errLeaseHeld) Error() string {
var errWakeTimeout = errors.New("timed out waiting for instance wake")
// Service is the activator (Eleven's activatorSvc). It structurally satisfies
-// the edge package's Activator interface — internal/edge is deliberately not
-// imported: the dependency points edge→activator only.
+// the edge package's Activator interface — the edge never imports this
+// package: the dependency points activator→edge only (backend_exec.go takes
+// edge.Cache and edge.ListenControl; serve calls edge.MarkAnswered), and the
+// seam back is the interface the edge declares for itself.
type Service struct {
cfg Config
backend Backend
@@ -1297,6 +1300,9 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout
return true // the sidecar hung up while queued on a wake
}
w.Header().Set("Retry-After", "5")
+ // Ours, not the app's: the edge must not file this 503 as an
+ // edge error (edge/answered.go; spec 2026-09-07 §3.2).
+ edge.MarkAnswered(r)
http.Error(w, "instance is asleep; sidecar requests do not wake it", http.StatusServiceUnavailable)
return true
}
@@ -1317,14 +1323,20 @@ func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Rout
case errors.As(err, &held):
retry := int(held.retryAfter/time.Second) + 1
w.Header().Set("Retry-After", strconv.Itoa(retry))
+ edge.MarkAnswered(r)
http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
case errors.Is(err, errWakeTimeout):
w.Header().Set("Retry-After", "5")
+ edge.MarkAnswered(r)
http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
case errors.Is(err, context.Canceled):
// The client hung up while queued; nothing to write.
default:
+ // The wake failure is already the member's platform log error
+ // (this log line, shipped by logsink.go); marking it keeps the
+ // edge from filing the same failure a second time.
log.Printf("activator: wake %s: %v", rt.Host, err)
+ edge.MarkAnswered(r)
http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
}
return true
diff --git a/internal/edge/answered.go b/internal/edge/answered.go
new file mode 100644
index 000000000..c74fbd578
--- /dev/null
+++ b/internal/edge/answered.go
@@ -0,0 +1,37 @@
+package edge
+
+import (
+ "context"
+ "net/http"
+)
+
+// answeredKey carries the flag the edge attaches so it can tell a 503 the
+// activator wrote on the wake path from one the upstream returned. The
+// wake failure already reaches the member as a platform log error
+// (emitEvent, logsink.go); the edge must not also file it as an edge
+// error (spec 2026-09-07 §3.2, "exactly once").
+//
+// The flag lives HERE, in the edge, rather than in the activator, because
+// the import direction between the two is already fixed the other way:
+// internal/activator imports internal/edge (backend_exec.go, for
+// edge.Cache and edge.ListenControl) while the edge reaches the activator
+// only through the Activator interface (activator.go's doc: "internal/edge
+// is deliberately not" imported). So the activator calls edge.MarkAnswered
+// before each of its four wake-path 503s; putting these two functions in
+// the activator instead would be an import cycle.
+type answeredKey struct{}
+
+// WithAnswerFlag returns r carrying a fresh flag and the flag itself.
+func WithAnswerFlag(r *http.Request) (*http.Request, *bool) {
+ f := new(bool)
+ return r.WithContext(context.WithValue(r.Context(), answeredKey{}, f)), f
+}
+
+// MarkAnswered sets the flag when r carries one; a no-op otherwise — so the
+// activator can call it unconditionally, including on the paths (a
+// non-hibernating route, a direct unit test) where the edge attached none.
+func MarkAnswered(r *http.Request) {
+ if f, ok := r.Context().Value(answeredKey{}).(*bool); ok {
+ *f = true
+ }
+}
diff --git a/internal/edge/answered_test.go b/internal/edge/answered_test.go
new file mode 100644
index 000000000..8c60a67e5
--- /dev/null
+++ b/internal/edge/answered_test.go
@@ -0,0 +1,19 @@
+package edge
+
+import (
+ "net/http/httptest"
+ "testing"
+)
+
+func TestAnswerFlagRoundTrip(t *testing.T) {
+ r := httptest.NewRequest("GET", "http://h/", nil)
+ MarkAnswered(r) // no flag: a no-op, never a panic
+ r, flag := WithAnswerFlag(r)
+ if *flag {
+ t.Fatal("flag must start false")
+ }
+ MarkAnswered(r)
+ if !*flag {
+ t.Fatal("MarkAnswered must set the flag the edge holds")
+ }
+}
diff --git a/internal/edge/edge.go b/internal/edge/edge.go
index a58f17548..1a372d47a 100644
--- a/internal/edge/edge.go
+++ b/internal/edge/edge.go
@@ -48,6 +48,7 @@ import (
"golang.org/x/crypto/acme/autocert"
+ "github.com/carlosframework/platform/internal/apperr"
"github.com/carlosframework/platform/internal/edgesfile"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -121,6 +122,9 @@ type Edge struct {
// tunnelLoop attaches via the ready hook — serving may already be
// underway.
usageMeter atomic.Pointer[UsageMeter]
+ // errSink, when set, receives edge-observed errors (an upstream 5xx)
+ // — spec 2026-09-07 §3.2. Nil disables emission; nothing else changes.
+ errSink atomic.Pointer[errorSinkRef]
// tunnels, when non-nil, is the reverse-channel hub (tunnel.go): remote
// hosts register on its reserved hostname and routes with Tunnel set are
// proxied over the channel they name. nil = tunnel serving not configured
@@ -321,6 +325,44 @@ func (e *Edge) EnableTunnels(hub *TunnelHub) {
// meter nothing. Safe to call after serving has started (atomic).
func (e *Edge) EnableUsage(m *UsageMeter) { e.usageMeter.Store(m) }
+// ErrorSink is where the edge files an error it observed on the serving
+// path. apperr.Sink satisfies it; the interface lives here so this package
+// depends on the contract, not the buffer.
+type ErrorSink interface {
+ Add(acct, app string, r apperr.Record)
+}
+
+type errorSinkRef struct{ ErrorSink }
+
+// EnableErrors wires the error sink in. Called once at startup by the
+// agent; nil is a valid no-op.
+func (e *Edge) EnableErrors(s ErrorSink) {
+ if s == nil {
+ e.errSink.Store(nil)
+ return
+ }
+ e.errSink.Store(&errorSinkRef{s})
+}
+
+// emitEdgeError files one edge-observed 5xx for rt. The path is sanitised
+// (query and token segments off) before it becomes the record's name.
+func (e *Edge) emitEdgeError(rt registry.Route, r *http.Request, status int) {
+ ref := e.errSink.Load()
+ if ref == nil {
+ return
+ }
+ path := apperr.SanitizePath(r.URL.Path)
+ ref.Add(release.RouteAccount(rt), rt.App, apperr.Record{
+ Host: rt.Host,
+ Source: apperr.SourceEdge,
+ Kind: apperr.KindHTTP,
+ Name: fmt.Sprintf("%d %s %s", status, r.Method, path),
+ Path: path,
+ Status: status,
+ Release: rt.Version,
+ })
+}
+
// armStoreDoor builds (or clears) the fleet store door from whatever is
// configured now. Called from both EnableStatic and EnableTunnels because the
// door needs one thing from each and the two are wired in either order at
@@ -1218,15 +1260,21 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
return
}
- // Meter every route this box serves except the platform's own status
- // host (not billable) — wrapping w here, once, before the upgrade/cache
- // branching below, is what makes all three dispatch shapes (hibernating,
- // cacheable, plain) plus the upgrade path count through this single
- // site: the defer fires after whichever branch actually served the
- // request. Deliberately placed after the tunnel-hostname and store-door
- // returns above (edge.go:531-544) — that hostname isn't a registry
- // route, so up.route isn't defined yet there, and neither one is
- // billable traffic.
+ // Observe every route this box serves except the platform's own status
+ // host (not billable, and not a member's app) — wrapping w here, once,
+ // before the upgrade/cache branching below, is what makes all three
+ // dispatch shapes (hibernating, cacheable, plain) plus the upgrade path
+ // pass through this single site: the defer fires after whichever branch
+ // actually served the request. Deliberately placed after the
+ // tunnel-hostname and store-door returns above (edge.go:531-544) — that
+ // hostname isn't a registry route, so up.route isn't defined yet there,
+ // and neither one is billable traffic.
+ //
+ // The observing writer is installed UNCONDITIONALLY now (spec
+ // 2026-09-07 §3.3) — the usage meter is one of its two consumers, the
+ // errors sink the other — because a box that meters nothing still has
+ // errors to file. Meter and sink are each nil-safe, so a pure-proxy
+ // `carlos edge` pays one wrapper and no more.
//
// hopped is the ATTRIBUTION half of the backhaul egress ruling
// (2026-08-24): a request that arrived over a verified hop is metered
@@ -1238,10 +1286,26 @@ func (e *Edge) ServeHTTP(w http.ResponseWriter, r *http.Request) {
// about the wrapper stack moves for this — recorder above the
// compressor, this counter below it — because the annotation changes
// WHAT is written at drain, never WHERE the bytes are counted.
- if m := e.usageMeter.Load(); m != nil && up.route.Kind != "status" {
- var bytes int64
- w = &countingWriter{ResponseWriter: w, n: &bytes}
- defer func() { m.Record(up.route, 1, bytes, hopped) }()
+ var bytes int64
+ ob := &countingWriter{ResponseWriter: w, n: &bytes}
+ w = ob
+ var answered *bool
+ if up.hibernating {
+ // The activator marks the wake-path 503s it writes itself
+ // (MarkAnswered, answered.go); those are already the member's
+ // platform log error and are not filed again as edge errors
+ // (spec §3.2, "exactly once").
+ r, answered = WithAnswerFlag(r)
+ }
+ if up.route.Kind != "status" {
+ defer func() {
+ if m := e.usageMeter.Load(); m != nil {
+ m.Record(up.route, 1, bytes, hopped)
+ }
+ if ob.status >= 500 && (answered == nil || !*answered) {
+ e.emitEdgeError(up.route, r, ob.status)
+ }
+ }()
}
// A Connection: Upgrade request (a WebSocket handshake, typically GET)
diff --git a/internal/edge/edgeerrors_test.go b/internal/edge/edgeerrors_test.go
new file mode 100644
index 000000000..f6c20bf5f
--- /dev/null
+++ b/internal/edge/edgeerrors_test.go
@@ -0,0 +1,113 @@
+package edge
+
+import (
+ "net/http"
+ "net/http/httptest"
+ "strings"
+ "sync"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/apperr"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type memErrSink struct {
+ mu sync.Mutex
+ recs []apperr.Record
+ apps []string
+}
+
+func (m *memErrSink) Add(acct, app string, r apperr.Record) {
+ m.mu.Lock()
+ defer m.mu.Unlock()
+ m.recs = append(m.recs, r)
+ m.apps = append(m.apps, acct+"/"+app)
+}
+
+func TestEdgeEmitsErrorOnUpstream5xx(t *testing.T) {
+ backend := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ switch r.URL.Path {
+ case "/boom":
+ http.Error(w, "kaboom", http.StatusInternalServerError)
+ case "/missing":
+ http.NotFound(w, r)
+ default:
+ w.Header().Set("Content-Type", "text/html")
+ w.Write([]byte("<p>ok</p>"))
+ }
+ }))
+ t.Cleanup(backend.Close)
+ reg := openReg(t)
+ rt := registry.Route{Host: "app.example.com", Kind: "service", Addr: backend.Listener.Addr().String(), Status: "ready", App: "web", Version: "v7"}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sink := &memErrSink{}
+ e.EnableErrors(sink)
+ srv := httptest.NewServer(e)
+ t.Cleanup(srv.Close)
+
+ do := func(method, path string) int {
+ req, _ := http.NewRequest(method, srv.URL+path, nil)
+ req.Host = "app.example.com"
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatal(err)
+ }
+ resp.Body.Close()
+ return resp.StatusCode
+ }
+ if do("POST", "/boom?token=abc") != 500 || do("GET", "/missing") != 404 || do("GET", "/") != 200 {
+ t.Fatal("upstream statuses must pass through unchanged")
+ }
+ if len(sink.recs) != 1 {
+ t.Fatalf("errors emitted = %d, want 1 (the 500 only)", len(sink.recs))
+ }
+ r := sink.recs[0]
+ if r.Source != apperr.SourceEdge || r.Kind != apperr.KindHTTP || r.Status != 500 || r.Host != "app.example.com" || r.Release != "v7" {
+ t.Fatalf("record = %+v", r)
+ }
+ if r.Name != "500 POST /boom" || r.Path != "/boom" || strings.Contains(r.Name, "token") {
+ t.Fatalf("name/path = %q %q", r.Name, r.Path)
+ }
+}
+
+// answering503 is an Activator that answers every request itself with the
+// wake-path 503 the real one writes, and marks it as the activator's own —
+// the same MarkAnswered call internal/activator's serve makes before each of
+// its four wake-path http.Error(…, 503) writes.
+type answering503 struct{}
+
+func (answering503) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
+ MarkAnswered(r)
+ http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
+ return true
+}
+func (answering503) RetireExcept(map[string]bool) {}
+
+func TestEdgeSkipsActivatorAnswered503(t *testing.T) {
+ reg := openReg(t)
+ rt := registry.Route{Host: "sleepy.example.com", Kind: "instance", Socket: t.TempDir() + "/x.sock", Status: "ready", Hibernate: true, App: "web"}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ e, err := New(reg, answering503{})
+ if err != nil {
+ t.Fatal(err)
+ }
+ sink := &memErrSink{}
+ e.EnableErrors(sink)
+ rec := httptest.NewRecorder()
+ req := httptest.NewRequest("GET", "http://sleepy.example.com/", nil)
+ e.ServeHTTP(rec, req)
+ if rec.Code != 503 {
+ t.Fatalf("status = %d", rec.Code)
+ }
+ if len(sink.recs) != 0 {
+ t.Fatalf("an activator-answered 503 is already a platform log error; got %d edge errors", len(sink.recs))
+ }
+}
diff --git a/internal/edge/usagemeter.go b/internal/edge/usagemeter.go
index 1cbd7db65..6afb2fa38 100644
--- a/internal/edge/usagemeter.go
+++ b/internal/edge/usagemeter.go
@@ -57,16 +57,45 @@ func (u *UsageMeter) Drain() map[string]usage.EdgeRow {
return out
}
-// countingWriter counts response body bytes. Deliberately does NOT
-// implement io.ReaderFrom: io.Copy falls back to Write, which counts.
-// Hijack/Flush forward via the interfaces the wrapped writer provides,
-// mirroring recordingResponseWriter (edge.go).
+// countingWriter counts response body bytes and OBSERVES the final status
+// and Content-Type (spec 2026-09-07 §3.3): it is the one wrapper installed
+// at a single site for every response shape — cache hit, plain proxy,
+// hibernating, upgrade — so it is where the edge learns that an upstream
+// answered 5xx (an edge error) or served HTML (a pageview, step 2). It
+// sits below the compressor, so Content-Type is the synced value
+// gzipWriter.syncHeader copied down before calling this WriteHeader.
+// Deliberately does NOT implement io.ReaderFrom: io.Copy falls back to
+// Write, which counts. Hijack/Flush forward via the interfaces the wrapped
+// writer provides, mirroring recordingResponseWriter (edge.go).
type countingWriter struct {
http.ResponseWriter
- n *int64
+ n *int64
+ status int // final status (>= 200); 0 until one is written
+ ctype string // Content-Type at the final WriteHeader
+ wroteHdr bool
+}
+
+// WriteHeader latches the first final status. 1xx are forwarded and never
+// latched (recordingResponseWriter.WriteHeader's reasoning).
+func (c *countingWriter) WriteHeader(status int) {
+ if status < 200 {
+ c.ResponseWriter.WriteHeader(status)
+ return
+ }
+ if !c.wroteHdr {
+ c.wroteHdr = true
+ c.status = status
+ c.ctype = c.ResponseWriter.Header().Get("Content-Type")
+ }
+ c.ResponseWriter.WriteHeader(status)
}
func (c *countingWriter) Write(p []byte) (int, error) {
+ if !c.wroteHdr {
+ c.wroteHdr = true
+ c.status = http.StatusOK
+ c.ctype = c.ResponseWriter.Header().Get("Content-Type")
+ }
n, err := c.ResponseWriter.Write(p)
*c.n += int64(n)
return n, err
diff --git a/internal/edge/usagemeter_test.go b/internal/edge/usagemeter_test.go
index abe4e46dd..15dbdb69b 100644
--- a/internal/edge/usagemeter_test.go
+++ b/internal/edge/usagemeter_test.go
@@ -90,3 +90,29 @@ func TestCountingWriterCountsBytes(t *testing.T) {
t.Fatalf("forwarding broken: %d %q", rec.Code, rec.Body.String())
}
}
+
+func TestCountingWriterObservesStatusAndType(t *testing.T) {
+ var n int64
+ rec := httptest.NewRecorder()
+ w := &countingWriter{ResponseWriter: rec, n: &n}
+ w.Header().Set("Content-Type", "text/html; charset=utf-8")
+ w.WriteHeader(103) // informational: forwarded, not latched
+ w.WriteHeader(http.StatusBadGateway)
+ w.WriteHeader(http.StatusOK) // superfluous second final status: ignored
+ if w.status != http.StatusBadGateway || w.ctype != "text/html; charset=utf-8" {
+ t.Fatalf("latched %d %q", w.status, w.ctype)
+ }
+ // Implicit 200 at first Write.
+ rec2 := httptest.NewRecorder()
+ w2 := &countingWriter{ResponseWriter: rec2, n: &n}
+ w2.Write([]byte("x"))
+ if w2.status != http.StatusOK {
+ t.Fatalf("implicit status = %d", w2.status)
+ }
+ // Nothing written at all: status stays 0 (the caller treats it as a
+ // bodiless 200 that is never a pageview and never an error).
+ w3 := &countingWriter{ResponseWriter: httptest.NewRecorder(), n: &n}
+ if w3.status != 0 {
+ t.Fatal("untouched writer must report 0")
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 04e45887d..14d1385a2 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -439,6 +439,13 @@ func RunAgent(args []string) {
setUpstreamSource(e.UpstreamStates)
m := edge.NewUsageMeter()
e.EnableUsage(m)
+ // Same arming point as the meter: the edge observes an upstream 5xx
+ // on the serving path and files it here (spec 2026-09-07 §3.2), and
+ // errshipLoop above ships what this fills. Guarded, because a typed
+ // nil inside the interface would NOT be nil to the edge.
+ if errSink != nil {
+ e.EnableErrors(errSink)
+ }
go usageLoop(m.Drain)
},
// Certificate issuance is a deployment-global duty, so it runs where