amadan

carlos / platform Public

tunnelclient: extract the host dial/serve/backoff loop into a leaf package

internal/host imports internal/edge, so internal/edge's tests can never
import internal/host — and the acceptance test for the reverse channel has
to drive the REAL client loop (backoff and all) or the reconnect bound it
asserts measures a test double. Move dial/serve/backoff into
internal/tunnelclient, stdlib + x/net only, and leave internal/host with the
part that is genuinely agent-shaped: env reading and publishing the state
the status heartbeat reports.

The hardening from the fix rounds moves across unchanged: the 10s handshake
deadline on both the dial and the /register round trip, the stable-uptime
gate on resetting backoff to the floor, and the unconditional jittered sleep
before every re-dial. Two adaptations: Loop and Dial take a context (so a
test can stop a fake host deterministically — cancellation closes the live
conn, which is the only stop signal http2.Server.ServeConn has), and the log
prefix moves to a Config.Logf the agent supplies, so the journal lines stay
byte-identical to what operators grep for.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 58f261ccfd22da2739b8cf7ae0729ca39db57854 parent 9f32baf
4 files changed, +450 −235
  • internal/host/tunnel_client.go +22 −160
  • internal/host/tunnel_client_test.go +14 −75
  • internal/tunnelclient/tunnelclient.go +260 −0
  • internal/tunnelclient/tunnelclient_test.go +154 −0
diff --git a/internal/host/tunnel_client.go b/internal/host/tunnel_client.go
index e5e38274b..555605a00 100644
--- a/internal/host/tunnel_client.go
+++ b/internal/host/tunnel_client.go
@@ -1,24 +1,20 @@
-// The host half of the reverse channel (issue #53 / C6 Step 3): dial OUT to
-// the edge, invert the connection, and serve our own local edge routing over
-// HTTP/2. Reconnect fast — hq's dual-WAN failover and a Pi's Wi-Fi blip are
-// both "the conn died, dial again", and that latency is user-visible.
+// The agent's wiring for the host half of the reverse channel (issue #53 / C6
+// Step 3). The dial/serve/backoff machinery itself lives in
+// internal/tunnelclient — a leaf package, so the edge's acceptance test can
+// drive the SAME loop this agent runs (internal/host imports internal/edge,
+// so the edge tests can never import this package). What is left here is the
+// part that is genuinely agent-shaped: reading the env, and publishing the
+// state the status heartbeat reports.
package host
import (
- "bufio"
- "crypto/tls"
- "fmt"
- "io"
+ "context"
"log"
- "math/rand/v2"
- "net"
"net/http"
- "net/url"
"os"
"sync/atomic"
- "time"
- "golang.org/x/net/http2"
+ "github.com/carlosframework/platform/internal/tunnelclient"
)
// tunnelStateVal is "" (disabled/never started), "connected", or
@@ -33,25 +29,6 @@ func tunnelState() string {
return ""
}
-const (
- tunnelBackoffFloor = 250 * time.Millisecond
- tunnelBackoffCap = 2 * time.Second
- // tunnelStableConnected is how long a connection has to have stayed up
- // before its loss resets backoff to the floor. Below this, the
- // disconnect is treated like a dial failure and backoff keeps
- // escalating: the edge's re-register-replaces-the-previous-channel
- // semantics mean two peers sharing a CARLOS_TUNNEL_NAME (a cloned Pi
- // env, the agent started twice, a promote/adopt/restart overlap window)
- // would otherwise evict each other in an unbounded zero-sleep loop.
- tunnelStableConnected = 5 * time.Second
- // tunnelHandshakeTimeout bounds dial+TLS (via net.Dialer /
- // tls.DialWithDialer) and, separately, the /register write+ReadResponse
- // round trip: an edge that accepts the TCP conn and then never answers
- // must not park tunnelLoop forever with no retry and no visible state
- // change.
- tunnelHandshakeTimeout = 10 * time.Second
-)
-
// tunnelLoop dials CARLOS_TUNNEL_URL and serves handler over the inverted
// conn, forever, with jittered capped backoff. Inert unless all three env
// vars are set — the same convention as every other optional agent loop.
@@ -66,132 +43,17 @@ func tunnelLoop(handler http.Handler) {
return
}
log.Printf("agent: tunnel loop armed (%s as %q)", rawURL, name)
- backoff := tunnelBackoffFloor
- for {
- // Set BEFORE dialing, not only on error: dialTunnel can block for up
- // to tunnelHandshakeTimeout on a black-holed handshake, and a stale
- // "" or "connected" during that window would read to Task 6's
- // heartbeat probe as either "never started" or "healthy" — both
- // wrong while we're actually hung waiting on a dead edge.
- tunnelStateVal.Store("reconnecting")
- conn, err := dialTunnel(rawURL, name, token)
- if err != nil {
- sleep := backoff + rand.N(backoff/2)
- log.Printf("agent: tunnel dial: %v (retry in %s)", err, sleep)
- time.Sleep(sleep)
- if backoff *= 2; backoff > tunnelBackoffCap {
- backoff = tunnelBackoffCap
- }
- continue
- }
- tunnelStateVal.Store("connected")
- log.Printf("agent: tunnel connected as %q", name)
- connectedAt := time.Now()
- serveTunnel(conn, handler) // blocks until the conn dies
- tunnelStateVal.Store("reconnecting")
- log.Printf("agent: tunnel lost — re-dialing")
- // A connection that stayed up a while earns a fast retry (the
- // dual-WAN-failover / Wi-Fi-blip case); one that dies instantly does
- // not — see tunnelStableConnected. Either way we always sleep some
- // amount before redialing: a zero-delay retry is exactly the storm
- // this guards against.
- if time.Since(connectedAt) >= tunnelStableConnected {
- backoff = tunnelBackoffFloor
- } else if backoff *= 2; backoff > tunnelBackoffCap {
- backoff = tunnelBackoffCap
- }
- sleep := backoff + rand.N(backoff/2)
- log.Printf("agent: tunnel retry in %s", sleep)
- time.Sleep(sleep)
- }
-}
-
-// bufferedConn replays bytes the handshake's bufio.Reader over-read (the h2
-// preface can arrive in the same packet as the 101) before the raw conn.
-type bufferedConn struct {
- r io.Reader
- net.Conn
-}
-
-func (b bufferedConn) Read(p []byte) (int, error) { return b.r.Read(p) }
-
-// dialTunnel 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.
-func dialTunnel(rawURL, name, token string) (net.Conn, error) {
- u, err := url.Parse(rawURL)
- if err != nil {
- return nil, fmt.Errorf("tunnel url: %w", err)
- }
- addr := u.Host
- dialer := &net.Dialer{Timeout: tunnelHandshakeTimeout}
- var conn net.Conn
- switch u.Scheme {
- case "https":
- if u.Port() == "" {
- addr = net.JoinHostPort(u.Hostname(), "443")
- }
- // DialWithDialer applies dialer.Timeout to the TCP connect and the
- // TLS handshake as a whole (crypto/tls docs) — a black-holed TLS
- // peer fails the dial instead of hanging it.
- conn, err = tls.DialWithDialer(dialer, "tcp", addr, &tls.Config{
- ServerName: u.Hostname(),
- NextProtos: []string{"http/1.1"}, // the handshake is 1.1; h2 starts after the 101
- })
- case "http":
- conn, err = dialer.Dial("tcp", addr)
- default:
- return nil, fmt.Errorf("tunnel url: unsupported scheme %q", u.Scheme)
- }
- if err != nil {
- return nil, err
- }
- // The dial above only bounds TCP connect + TLS handshake; the
- // /register write+read is a separate hang risk (an edge that accepts
- // the conn and then never answers). Bound it too, and clear the
- // deadline before returning success — the live h2 session that follows
- // must not inherit a stale deadline from the handshake.
- if err := conn.SetDeadline(time.Now().Add(tunnelHandshakeTimeout)); err != nil {
- conn.Close()
- return nil, fmt.Errorf("tunnel: set handshake deadline: %w", err)
- }
- req := &http.Request{
- Method: "GET",
- URL: &url.URL{Path: "/register"},
- Host: u.Hostname(),
- Header: http.Header{
- "Authorization": {"Bearer " + token},
- "Carlos-Tunnel-Name": {name},
- "Connection": {"Upgrade"},
- "Upgrade": {"carlos-tunnel/1"},
- },
- }
- if err := req.Write(conn); err != nil {
- conn.Close()
- return nil, err
- }
- br := bufio.NewReader(conn)
- resp, err := http.ReadResponse(br, req)
- if err != nil {
- conn.Close()
- return nil, err
- }
- resp.Body.Close()
- if resp.StatusCode != http.StatusSwitchingProtocols {
- conn.Close()
- return nil, fmt.Errorf("tunnel register: %s", resp.Status)
- }
- if err := conn.SetDeadline(time.Time{}); err != nil {
- conn.Close()
- return nil, fmt.Errorf("tunnel: clear handshake deadline: %w", err)
- }
- if br.Buffered() > 0 {
- return bufferedConn{r: io.MultiReader(br, conn), Conn: conn}, nil
- }
- return conn, nil
-}
-
-// serveTunnel serves handler over the inverted conn until it dies.
-func serveTunnel(conn net.Conn, handler http.Handler) {
- (&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: handler})
+ // context.Background(): the agent's tunnel runs for the life of the
+ // process and is torn down with it. The ctx exists for callers that need
+ // to stop the loop deterministically (the edge's e2e), not for this one.
+ tunnelclient.Loop(context.Background(), tunnelclient.Config{
+ URL: rawURL,
+ Name: name,
+ Token: token,
+ Handler: handler,
+ OnState: func(s string) { tunnelStateVal.Store(s) },
+ // The "agent: " prefix is what operators grep the journal for, so it
+ // is preserved here rather than baked into the leaf package.
+ Logf: func(format string, args ...any) { log.Printf("agent: "+format, args...) },
+ })
}
diff --git a/internal/host/tunnel_client_test.go b/internal/host/tunnel_client_test.go
index b9315999a..ed8904d87 100644
--- a/internal/host/tunnel_client_test.go
+++ b/internal/host/tunnel_client_test.go
@@ -2,85 +2,24 @@ package host
import (
"fmt"
- "io"
"net/http"
"net/http/httptest"
"testing"
"time"
-
- "golang.org/x/net/http2"
)
-// fakeTunnelEdge accepts one registration, answers 101, then drives ONE h2
-// request over the inverted conn and returns the response body it read.
-func fakeTunnelEdge(t *testing.T, wantToken, wantName string) (*httptest.Server, chan string) {
- t.Helper()
- got := make(chan string, 1)
- srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- if r.Header.Get("Authorization") != "Bearer "+wantToken ||
- r.Header.Get("Carlos-Tunnel-Name") != wantName {
- http.Error(w, "unauthorized", http.StatusUnauthorized)
- return
- }
- conn, bufrw, err := w.(http.Hijacker).Hijack()
- if err != nil {
- t.Errorf("hijack: %v", err)
- return
- }
- fmt.Fprintf(bufrw, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: carlos-tunnel/1\r\nConnection: Upgrade\r\n\r\n")
- bufrw.Flush()
- cc, err := (&http2.Transport{}).NewClientConn(conn)
- if err != nil {
- t.Errorf("h2 client: %v", err)
- return
- }
- req, _ := http.NewRequest("GET", "https://demo.example/hello", nil)
- resp, err := cc.RoundTrip(req)
- if err != nil {
- t.Errorf("roundtrip over tunnel: %v", err)
- got <- ""
- return
- }
- body, _ := io.ReadAll(resp.Body)
- got <- resp.Header.Get("X-From-Host") + ":" + string(body)
- }))
- return srv, got
-}
-
-func TestDialAndServeInvertsConnection(t *testing.T) {
- srv, got := fakeTunnelEdge(t, "sekrit", "pi-box")
- defer srv.Close()
- handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- w.Header().Set("X-From-Host", r.Host)
- fmt.Fprint(w, "pong")
- })
- conn, err := dialTunnel(srv.URL, "pi-box", "sekrit")
- if err != nil {
- t.Fatalf("dialTunnel: %v", err)
- }
- go serveTunnel(conn, handler)
- if v := <-got; v != "demo.example:pong" {
- t.Fatalf("over-tunnel response = %q, want demo.example:pong", v)
- }
-}
-
-func TestDialTunnelRejectsNon101(t *testing.T) {
- srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
- http.Error(w, "unauthorized", http.StatusUnauthorized)
- }))
- defer srv.Close()
- if _, err := dialTunnel(srv.URL, "pi-box", "wrong"); err == nil {
- t.Fatal("want error on 401 handshake")
- }
-}
+// The dial/serve/handshake tests for this channel live in
+// internal/tunnelclient, where that code now lives. What is testable HERE is
+// the agent's wrapper: the env convention, and the state transitions the
+// status heartbeat reads off tunnelState().
-// TestTunnelLoopStateTransitions drives tunnelLoop (not dialTunnel/
-// serveTunnel directly) against a fake edge that hijacks, answers 101, then
-// holds the raw conn open until the test releases it — enough to observe
-// tunnelState() go "" -> "connected", and then "connected" -> "reconnecting"
-// once the conn is torn down. This is the state Task 6's heartbeat probe
-// reads, so its transitions are worth asserting directly, not just inferring
-// from dialTunnel/serveTunnel behavior.
+// TestTunnelLoopStateTransitions drives tunnelLoop against a fake edge that
+// hijacks, answers 101, then holds the raw conn open until the test releases
+// it — enough to observe tunnelState() go "" -> "connected", and then
+// "connected" -> "reconnecting" once the conn is torn down. This is the state
+// Task 6's heartbeat probe reads, so its transitions are worth asserting
+// through the wrapper that publishes them, not only in the package that
+// reports them.
func TestTunnelLoopStateTransitions(t *testing.T) {
release := make(chan struct{})
srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
@@ -101,9 +40,9 @@ func TestTunnelLoopStateTransitions(t *testing.T) {
t.Setenv("CARLOS_TUNNEL_NAME", "pi-box")
// tunnelLoop never returns while configured (by design — it retries
- // forever), so this goroutine outlives the test; the fake edge is torn
- // down via defer, and any further redial attempts just fail fast against
- // a closed listener.
+ // forever on context.Background()), so this goroutine outlives the test;
+ // the fake edge is torn down via defer, and any further redial attempts
+ // just fail fast against a closed listener.
go tunnelLoop(http.NewServeMux())
waitForTunnelState(t, "connected")
diff --git a/internal/tunnelclient/tunnelclient.go b/internal/tunnelclient/tunnelclient.go
new file mode 100644
index 000000000..f69b0c125
--- /dev/null
+++ b/internal/tunnelclient/tunnelclient.go
@@ -0,0 +1,260 @@
+// Package tunnelclient is the host half of the reverse channel (issue #53 /
+// C6 Step 3): dial OUT to the edge, invert the connection, and serve a local
+// http.Handler over HTTP/2 on it. Reconnect fast — hq's dual-WAN failover and
+// a Pi's Wi-Fi blip are both "the conn died, dial again", and that latency is
+// user-visible.
+//
+// It is deliberately a LEAF package: stdlib plus golang.org/x/net only, no
+// CARLOS imports. internal/host imports internal/edge, so the edge package's
+// own tests cannot import internal/host — and the acceptance test for this
+// channel (internal/edge/tunnel_e2e_test.go) has to drive the REAL client
+// loop, backoff and all, or the reconnect bound it asserts would be measuring
+// a test double. Keeping dial/serve/backoff here is what lets both sides use
+// one implementation. internal/host/tunnel_client.go is now a thin wrapper:
+// it reads the env and stores the state this package reports.
+package tunnelclient
+
+import (
+ "bufio"
+ "context"
+ "crypto/tls"
+ "fmt"
+ "io"
+ "log"
+ "math/rand/v2"
+ "net"
+ "net/http"
+ "net/url"
+ "time"
+
+ "golang.org/x/net/http2"
+)
+
+// The two states Loop reports through Config.OnState. "" (never reported
+// here) means the loop is not running at all — probes must fail toward honest
+// values, so the caller's zero value has to be distinguishable from anything
+// this package says.
+const (
+ StateConnected = "connected"
+ StateReconnecting = "reconnecting"
+)
+
+const (
+ BackoffFloor = 250 * time.Millisecond
+ BackoffCap = 2 * time.Second
+ // StableConnected is how long a connection has to have stayed up before
+ // its loss resets backoff to the floor. Below this, the disconnect is
+ // treated like a dial failure and backoff keeps escalating: the edge's
+ // re-register-replaces-the-previous-channel semantics mean two peers
+ // sharing a channel name (a cloned Pi env, the agent started twice, a
+ // promote/adopt/restart overlap window) would otherwise evict each other
+ // in an unbounded zero-sleep loop.
+ StableConnected = 5 * time.Second
+ // HandshakeTimeout bounds dial+TLS (via net.Dialer / tls.Dialer) and,
+ // separately, the /register write+ReadResponse round trip: an edge that
+ // accepts the TCP conn and then never answers must not park Loop forever
+ // with no retry and no visible state change.
+ HandshakeTimeout = 10 * time.Second
+)
+
+// Config is one channel's worth of dialing instructions. URL, Name, Token and
+// Handler are required — Loop does not police them, because the caller is the
+// one holding the config source and owes its user the specific complaint (see
+// internal/host's env check). A blank or unparseable URL simply fails every
+// dial and backs off to BackoffCap, which is a bad config loudly logged, not
+// a storm.
+type Config struct {
+ URL string // the edge's registration endpoint, http:// or https://
+ Name string // this channel's name, as the edge has it configured
+ Token string // the bearer token bound to Name
+ Handler http.Handler // served over the inverted conn
+
+ // OnState, when non-nil, is called with StateConnected /
+ // StateReconnecting on every transition — synchronously, from Loop's
+ // goroutine, so it must not block.
+ OnState func(string)
+ // Logf defaults to log.Printf. The agent passes one that prefixes
+ // "agent: " so the journal lines stay exactly what operators grep for.
+ Logf func(format string, args ...any)
+}
+
+// Loop dials cfg.URL and serves cfg.Handler over the inverted conn, forever,
+// with jittered capped backoff. It returns only when ctx is cancelled; a
+// cancelled ctx tears down the live conn rather than waiting for it to die on
+// its own. State is left at whatever the last transition reported — this
+// package never invents a "stopped" state, because the agent's heartbeat
+// probe reads "" as "the loop was never started", which is a different fact.
+func Loop(ctx context.Context, cfg Config) {
+ logf := cfg.Logf
+ if logf == nil {
+ logf = log.Printf
+ }
+ setState := func(s string) {
+ if cfg.OnState != nil {
+ cfg.OnState(s)
+ }
+ }
+ backoff := BackoffFloor
+ for ctx.Err() == nil {
+ // Set BEFORE dialing, not only on error: Dial can block for up to
+ // HandshakeTimeout on a black-holed handshake, and a stale "" or
+ // "connected" during that window would read to the agent's heartbeat
+ // probe as either "never started" or "healthy" — both wrong while
+ // we're actually hung waiting on a dead edge.
+ setState(StateReconnecting)
+ conn, err := Dial(ctx, cfg.URL, cfg.Name, cfg.Token)
+ if err != nil {
+ sleep := backoff + rand.N(backoff/2)
+ logf("tunnel dial: %v (retry in %s)", err, sleep)
+ if !wait(ctx, sleep) {
+ return
+ }
+ if backoff *= 2; backoff > BackoffCap {
+ backoff = BackoffCap
+ }
+ continue
+ }
+ setState(StateConnected)
+ logf("tunnel connected as %q", cfg.Name)
+ connectedAt := time.Now()
+ serveUntilDead(ctx, conn, cfg.Handler) // blocks until the conn dies
+ setState(StateReconnecting)
+ logf("tunnel lost — re-dialing")
+ // A connection that stayed up a while earns a fast retry (the
+ // dual-WAN-failover / Wi-Fi-blip case); one that dies instantly does
+ // not — see StableConnected. Either way we always sleep some amount
+ // before redialing: a zero-delay retry is exactly the storm this
+ // guards against.
+ if time.Since(connectedAt) >= StableConnected {
+ backoff = BackoffFloor
+ } else if backoff *= 2; backoff > BackoffCap {
+ backoff = BackoffCap
+ }
+ sleep := backoff + rand.N(backoff/2)
+ logf("tunnel retry in %s", sleep)
+ if !wait(ctx, sleep) {
+ return
+ }
+ }
+}
+
+// wait sleeps d, or returns false if ctx is cancelled first. Every sleep in
+// Loop goes through it: a loop that ignored cancellation would outlive the
+// test that started it and keep dialing a closed listener.
+func wait(ctx context.Context, d time.Duration) bool {
+ t := time.NewTimer(d)
+ defer t.Stop()
+ select {
+ case <-ctx.Done():
+ return false
+ case <-t.C:
+ return true
+ }
+}
+
+// serveUntilDead is Serve plus cancellation: http2.Server.ServeConn blocks
+// until the conn dies and has no other stop signal, so closing the conn is
+// how a cancelled ctx gets Loop back.
+func serveUntilDead(ctx context.Context, conn net.Conn, handler http.Handler) {
+ done := make(chan struct{})
+ go func() {
+ select {
+ case <-ctx.Done():
+ conn.Close()
+ case <-done:
+ }
+ }()
+ defer close(done)
+ Serve(conn, handler)
+}
+
+// bufferedConn replays bytes the handshake's bufio.Reader over-read (the h2
+// preface can arrive in the same packet as the 101) before the raw conn.
+type bufferedConn struct {
+ r io.Reader
+ net.Conn
+}
+
+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.
+func Dial(ctx context.Context, rawURL, name, token string) (net.Conn, error) {
+ u, err := url.Parse(rawURL)
+ if err != nil {
+ return nil, fmt.Errorf("tunnel url: %w", err)
+ }
+ addr := u.Host
+ dialer := &net.Dialer{Timeout: HandshakeTimeout}
+ var conn net.Conn
+ switch u.Scheme {
+ case "https":
+ if u.Port() == "" {
+ addr = net.JoinHostPort(u.Hostname(), "443")
+ }
+ // tls.Dialer applies dialer.Timeout to the TCP connect and the TLS
+ // handshake as a whole (crypto/tls docs) — a black-holed TLS peer
+ // fails the dial instead of hanging it — and ctx on top of that.
+ td := &tls.Dialer{NetDialer: dialer, Config: &tls.Config{
+ ServerName: u.Hostname(),
+ NextProtos: []string{"http/1.1"}, // the handshake is 1.1; h2 starts after the 101
+ }}
+ conn, err = td.DialContext(ctx, "tcp", addr)
+ case "http":
+ conn, err = dialer.DialContext(ctx, "tcp", addr)
+ default:
+ return nil, fmt.Errorf("tunnel url: unsupported scheme %q", u.Scheme)
+ }
+ if err != nil {
+ return nil, err
+ }
+ // The dial above only bounds TCP connect + TLS handshake; the /register
+ // write+read is a separate hang risk (an edge that accepts the conn and
+ // then never answers). Bound it too, and clear the deadline before
+ // returning success — the live h2 session that follows must not inherit a
+ // stale deadline from the handshake.
+ if err := conn.SetDeadline(time.Now().Add(HandshakeTimeout)); err != nil {
+ conn.Close()
+ return nil, fmt.Errorf("tunnel: set handshake deadline: %w", err)
+ }
+ req := &http.Request{
+ Method: "GET",
+ URL: &url.URL{Path: "/register"},
+ Host: u.Hostname(),
+ Header: http.Header{
+ "Authorization": {"Bearer " + token},
+ "Carlos-Tunnel-Name": {name},
+ "Connection": {"Upgrade"},
+ "Upgrade": {"carlos-tunnel/1"},
+ },
+ }
+ if err := req.Write(conn); err != nil {
+ conn.Close()
+ return nil, err
+ }
+ br := bufio.NewReader(conn)
+ resp, err := http.ReadResponse(br, req)
+ if err != nil {
+ conn.Close()
+ return nil, err
+ }
+ resp.Body.Close()
+ if resp.StatusCode != http.StatusSwitchingProtocols {
+ conn.Close()
+ return nil, fmt.Errorf("tunnel register: %s", resp.Status)
+ }
+ if err := conn.SetDeadline(time.Time{}); err != nil {
+ conn.Close()
+ return nil, fmt.Errorf("tunnel: clear handshake deadline: %w", err)
+ }
+ if br.Buffered() > 0 {
+ return bufferedConn{r: io.MultiReader(br, conn), Conn: conn}, nil
+ }
+ return conn, nil
+}
+
+// Serve serves handler over the inverted conn until it dies.
+func Serve(conn net.Conn, handler http.Handler) {
+ (&http2.Server{}).ServeConn(conn, &http2.ServeConnOpts{Handler: handler})
+}
diff --git a/internal/tunnelclient/tunnelclient_test.go b/internal/tunnelclient/tunnelclient_test.go
new file mode 100644
index 000000000..64e448cb5
--- /dev/null
+++ b/internal/tunnelclient/tunnelclient_test.go
@@ -0,0 +1,154 @@
+package tunnelclient
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+ "time"
+
+ "golang.org/x/net/http2"
+)
+
+// fakeTunnelEdge accepts one registration, answers 101, then drives ONE h2
+// request over the inverted conn and returns the response body it read.
+func fakeTunnelEdge(t *testing.T, wantToken, wantName string) (*httptest.Server, chan string) {
+ t.Helper()
+ got := make(chan string, 1)
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ if r.Header.Get("Authorization") != "Bearer "+wantToken ||
+ r.Header.Get("Carlos-Tunnel-Name") != wantName {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ return
+ }
+ conn, bufrw, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ t.Errorf("hijack: %v", err)
+ return
+ }
+ fmt.Fprintf(bufrw, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: carlos-tunnel/1\r\nConnection: Upgrade\r\n\r\n")
+ bufrw.Flush()
+ cc, err := (&http2.Transport{}).NewClientConn(conn)
+ if err != nil {
+ t.Errorf("h2 client: %v", err)
+ return
+ }
+ req, _ := http.NewRequest("GET", "https://demo.example/hello", nil)
+ resp, err := cc.RoundTrip(req)
+ if err != nil {
+ t.Errorf("roundtrip over tunnel: %v", err)
+ got <- ""
+ return
+ }
+ body, _ := io.ReadAll(resp.Body)
+ got <- resp.Header.Get("X-From-Host") + ":" + string(body)
+ }))
+ return srv, got
+}
+
+func TestDialAndServeInvertsConnection(t *testing.T) {
+ srv, got := fakeTunnelEdge(t, "sekrit", "pi-box")
+ defer srv.Close()
+ handler := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ w.Header().Set("X-From-Host", r.Host)
+ fmt.Fprint(w, "pong")
+ })
+ conn, err := Dial(context.Background(), srv.URL, "pi-box", "sekrit")
+ if err != nil {
+ t.Fatalf("Dial: %v", err)
+ }
+ go Serve(conn, handler)
+ if v := <-got; v != "demo.example:pong" {
+ t.Fatalf("over-tunnel response = %q, want demo.example:pong", v)
+ }
+}
+
+func TestDialRejectsNon101(t *testing.T) {
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ http.Error(w, "unauthorized", http.StatusUnauthorized)
+ }))
+ defer srv.Close()
+ if _, err := Dial(context.Background(), srv.URL, "pi-box", "wrong"); err == nil {
+ t.Fatal("want error on 401 handshake")
+ }
+}
+
+// A cancelled ctx must stop the loop from BOTH of its blocking positions:
+// parked in backoff between dials (no edge at all), and parked in ServeConn on
+// a live channel that is perfectly healthy. The second one is the load-bearing
+// case — http2.Server.ServeConn has no stop signal of its own, so cancellation
+// has to reach it by closing the conn — and it is what lets a test tear the
+// fake host down deterministically instead of leaking a dialer for the rest of
+// the run.
+func TestLoopStopsOnContextCancel(t *testing.T) {
+ t.Run("while backing off", func(t *testing.T) {
+ // A server that is already gone: every dial is refused, so the loop
+ // spends essentially all of its time parked in wait().
+ dead := httptest.NewServer(http.NotFoundHandler())
+ url := dead.URL
+ dead.Close()
+ assertLoopReturns(t, url, nil)
+ })
+ t.Run("while connected", func(t *testing.T) {
+ release := make(chan struct{})
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ conn, bufrw, err := w.(http.Hijacker).Hijack()
+ if err != nil {
+ t.Errorf("hijack: %v", err)
+ return
+ }
+ fmt.Fprintf(bufrw, "HTTP/1.1 101 Switching Protocols\r\nUpgrade: carlos-tunnel/1\r\nConnection: Upgrade\r\n\r\n")
+ bufrw.Flush()
+ // Hold the channel open from this side: the edge never hangs up,
+ // so the ONLY thing that can end ServeConn is cancellation
+ // closing the conn — which is the property under test.
+ <-release
+ conn.Close()
+ }))
+ // LIFO: release the fake edge's handler first, then shut it down.
+ defer srv.Close()
+ defer close(release)
+ assertLoopReturns(t, srv.URL, make(chan struct{}, 1))
+ })
+}
+
+// assertLoopReturns runs Loop against url, waits for it to reach the state
+// gate (connected, when connected is non-nil), cancels, and requires a prompt
+// return.
+func assertLoopReturns(t *testing.T, url string, connected chan struct{}) {
+ t.Helper()
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() {
+ defer close(done)
+ Loop(ctx, Config{
+ URL: url, Name: "pi-box", Token: "sekrit",
+ Handler: http.NewServeMux(),
+ OnState: func(s string) {
+ if s == StateConnected && connected != nil {
+ select {
+ case connected <- struct{}{}:
+ default:
+ }
+ }
+ },
+ Logf: func(string, ...any) {}, // the failing-dial case is noisy
+ })
+ }()
+ if connected != nil {
+ select {
+ case <-connected:
+ case <-time.After(5 * time.Second):
+ cancel()
+ t.Fatal("loop never connected")
+ }
+ }
+ cancel()
+ select {
+ case <-done:
+ case <-time.After(5 * time.Second):
+ t.Fatal("Loop did not return after its context was cancelled")
+ }
+}