amadan

carlos / platform Public

edge: e2e reverse-channel acceptance — round-trip, header discipline, reconnect bound

Issue #53's acceptance test, in-repo, loopback only: a real edge and a fake
remote host talking over a real inverted h2 channel. The fake host is not a
hand-rolled dialer — it runs internal/tunnelclient.Loop, the same loop
`carlos agent` runs on a Pi, so the reconnect bound is measured against the
production backoff rather than a test double.

What it pins:

  - one listener carries both app traffic and the tunnel registration (the
    no-new-port property);
  - a response's headers survive the channel, and Carlos-Purge does NOT —
    purge discipline holds over a tunnel exactly as it does over a socket;
  - Host and X-Forwarded-For arrive at the remote app the way they arrive at
    a local one;
  - a channel killed with a request PARKED in the remote handler fails that
    request fast instead of hanging it, the host re-dials, and the next
    request succeeds — all inside e2eReconnectBound (5s; the measured figure
    is ~0.65s, one jittered backoff step).

The kill is deterministic, not a sleep: the test waits for the remote
handler to signal it is in flight, then unregisters the channel under the
hub lock BEFORE closing the conn, so the host's re-registration can never
race the teardown.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 8064cfa1c622c71c86bb40d2352806d90fccc703 parent 58f261c
2 files changed, +225 −0
  • internal/edge/tunnel_e2e_test.go +218 −0
  • internal/tunnelclient/tunnelclient.go +7 −0
diff --git a/internal/edge/tunnel_e2e_test.go b/internal/edge/tunnel_e2e_test.go
new file mode 100644
index 000000000..5aa4ef5a4
--- /dev/null
+++ b/internal/edge/tunnel_e2e_test.go
@@ -0,0 +1,218 @@
+package edge
+
+// The acceptance test issue #53 names: edge + fake remote host over a REAL
+// reverse channel on loopback. Round-trip, header survival, kill-mid-flight,
+// re-dial, and the reconnect bound (asserted, not narrated).
+//
+// The fake remote host is not a hand-rolled dialer: it runs
+// internal/tunnelclient.Loop, the exact loop `carlos agent` runs on a Pi
+// (internal/host/tunnel_client.go is a thin env-reading wrapper around it).
+// That is the whole point of the extraction — a reconnect bound measured
+// against a test double would prove nothing about the fleet.
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "net/url"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/tunnelclient"
+)
+
+// e2eReconnectBound is the promise: a channel that dies is back and serving
+// within this. The client sleeps one jittered backoff step (250-375ms on a
+// first loss, 500-750ms once it has doubled) before re-dialing, so the real
+// figure is well under a second — the bound is deliberately loose enough that
+// a loaded CI box cannot flake it, and tight enough that a regression to the
+// 2s cap compounding, or to no reconnect at all, fails here.
+const e2eReconnectBound = 5 * time.Second
+
+func TestTunnelEndToEnd(t *testing.T) {
+ reg := tunnelRoute(t, registry.Route{
+ Host: "demo.example", Kind: "instance", Status: "ready", Tunnel: "pi-box",
+ })
+ e, err := New(reg, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ // The edge listens on loopback; both app traffic and the tunnel
+ // registration ride this ONE listener — the no-new-port property.
+ srv := httptest.NewServer(e)
+ defer srv.Close()
+
+ // The hub's reserved hostname has to be the name the host actually dials,
+ // because the dialer sends the URL's host as its Host header and the edge
+ // routes registration on that. On a loopback edge that name is the
+ // listener's own address; in production it is the operator's
+ // CARLOS_TUNNEL_HOST, resolved by DNS to the same 443.
+ u, err := url.Parse(srv.URL)
+ if err != nil {
+ t.Fatal(err)
+ }
+ hub := NewTunnelHub(u.Hostname(), map[string]string{"pi-box": "sekrit"})
+ e.EnableTunnels(hub)
+ if err := e.Reload(); err != nil {
+ t.Fatal(err)
+ }
+
+ // Fake remote host: its "local edge" answers with a header + body, and
+ // echoes Host and X-Forwarded-For so the test can prove header discipline
+ // held all the way across the channel. /slow parks until released — that
+ // is what makes the kill below genuinely mid-flight rather than a sleep
+ // and a hope.
+ inflight := make(chan struct{}, 1)
+ release := make(chan struct{})
+ defer close(release)
+ appHandler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.URL.Path == "/slow" {
+ select {
+ case inflight <- struct{}{}:
+ default:
+ }
+ // Either the test lets it finish, or the stream dies under it —
+ // which is what a real handler sees when the channel drops, and
+ // what keeps this goroutine from outliving the connection.
+ select {
+ case <-release:
+ case <-r.Context().Done():
+ }
+ }
+ w.Header().Set("X-App-Header", "survived")
+ w.Header().Set("Carlos-Purge", "/should-be-stripped") // edge must strip
+ fmt.Fprintf(w, "host=%s xff=%s", r.Host, r.Header.Get("X-Forwarded-For"))
+ })
+ stop := startFakeRemoteHost(t, srv.URL, "pi-box", "sekrit", appHandler)
+ defer stop()
+
+ waitForTunnelState(t, hub, "pi-box", "connected", e2eReconnectBound)
+
+ do := func(path string) (*http.Response, string) {
+ t.Helper()
+ req, err := http.NewRequest("GET", srv.URL+path, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.Host = "demo.example"
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ t.Fatalf("request through tunnel: %v", err)
+ }
+ body, _ := io.ReadAll(resp.Body)
+ resp.Body.Close()
+ return resp, string(body)
+ }
+
+ resp, body := do("/")
+ if resp.StatusCode != 200 || resp.Header.Get("X-App-Header") != "survived" {
+ t.Fatalf("round-trip: %d %v", resp.StatusCode, resp.Header)
+ }
+ if resp.Header.Get("Carlos-Purge") != "" {
+ t.Fatal("Carlos-Purge leaked to the client — purge discipline broken over tunnel")
+ }
+ if !strings.Contains(body, "host=demo.example") || !strings.Contains(body, "xff=127.") {
+ t.Fatalf("host/XFF discipline broken: %q", body)
+ }
+
+ // Kill the channel out from under the host with a request PARKED inside
+ // the remote handler, then assert three things: the in-flight request
+ // fails fast rather than hanging, the host re-dials, and the next request
+ // succeeds — all inside the bound.
+ slowDone := make(chan int, 1)
+ go func() {
+ req, err := http.NewRequest("GET", srv.URL+"/slow", nil)
+ if err != nil {
+ slowDone <- -1
+ return
+ }
+ req.Host = "demo.example"
+ resp, err := http.DefaultClient.Do(req)
+ if err != nil {
+ slowDone <- 0 // the client saw the failure directly
+ return
+ }
+ resp.Body.Close()
+ slowDone <- resp.StatusCode
+ }()
+ select {
+ case <-inflight:
+ case <-time.After(e2eReconnectBound):
+ t.Fatal("/slow never reached the remote host's handler")
+ }
+
+ // Unregister BEFORE closing, so the ordering is deterministic rather than
+ // a race with the host's re-registration: once the entry is gone, State is
+ // "" until the NEW channel lands, and a re-register can never be clobbered
+ // by this teardown.
+ hub.mu.Lock()
+ dead := hub.conns["pi-box"]
+ delete(hub.conns, "pi-box")
+ hub.mu.Unlock()
+ if dead == nil {
+ t.Fatal("no live channel to kill — the hub lost it before the test did")
+ }
+ start := time.Now()
+ dead.raw.Close()
+
+ select {
+ case code := <-slowDone:
+ if code < 0 {
+ t.Fatal("could not build the in-flight request")
+ }
+ if code == http.StatusOK {
+ t.Fatal("the in-flight request succeeded after its channel died — the edge served bytes over a dead channel")
+ }
+ case <-time.After(e2eReconnectBound):
+ t.Fatal("in-flight request neither completed nor failed after its channel died — it hung")
+ }
+
+ waitForTunnelState(t, hub, "pi-box", "connected", e2eReconnectBound)
+ if resp, _ := do("/"); resp.StatusCode != 200 {
+ t.Fatalf("post-reconnect request: %d", resp.StatusCode)
+ }
+ if d := time.Since(start); d > e2eReconnectBound {
+ t.Fatalf("reconnect took %s, bound is %s", d, e2eReconnectBound)
+ } else {
+ t.Logf("reconnected and served in %s (bound %s)", d, e2eReconnectBound)
+ }
+}
+
+// startFakeRemoteHost runs the production host-side loop against edgeURL and
+// returns a stop func that cancels it and WAITS for it to exit — a loop still
+// dialing after its test returned would log into the next test's output and
+// hold a goroutine for the rest of the run.
+func startFakeRemoteHost(t *testing.T, edgeURL, name, token string, handler http.Handler) func() {
+ t.Helper()
+ ctx, cancel := context.WithCancel(context.Background())
+ var wg sync.WaitGroup
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ tunnelclient.Loop(ctx, tunnelclient.Config{
+ URL: edgeURL, Name: name, Token: token, Handler: handler,
+ Logf: func(format string, args ...any) { t.Logf("fake host: "+format, args...) },
+ })
+ }()
+ return func() {
+ cancel()
+ wg.Wait() // no t.Logf after the test returns
+ }
+}
+
+func waitForTunnelState(t *testing.T, hub *TunnelHub, name, want string, within time.Duration) {
+ t.Helper()
+ deadline := time.Now().Add(within)
+ for time.Now().Before(deadline) {
+ if hub.State(name) == want {
+ return
+ }
+ time.Sleep(20 * time.Millisecond)
+ }
+ t.Fatalf("tunnel %s never reached state %q within %s", name, want, within)
+}
diff --git a/internal/tunnelclient/tunnelclient.go b/internal/tunnelclient/tunnelclient.go
index f69b0c125..92b4680f6 100644
--- a/internal/tunnelclient/tunnelclient.go
+++ b/internal/tunnelclient/tunnelclient.go
@@ -180,6 +180,13 @@ func (b bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) }
// Dial performs the registration handshake and returns the raw conn, ready to
// serve h2 on. https URLs use real TLS (the edge's public cert — server auth
// by PKI, client auth by bearer token); http is for tests.
+//
+// ctx bounds the TCP connect and the TLS handshake. The /register round trip
+// after that is bounded by HandshakeTimeout instead, not by ctx: a cancel
+// arriving mid-handshake is therefore observed within HandshakeTimeout rather
+// than immediately, which is the one place cancellation is not instant and is
+// deliberately left simple — the alternative is a watchdog goroutine per dial
+// to buy 10 seconds on a shutdown path.
func Dial(ctx context.Context, rawURL, name, token string) (net.Conn, error) {
u, err := url.Parse(rawURL)
if err != nil {