amadan

carlos / platform Public

host: tunnel dial-out loop — reverse channel client with capped jittered backoff

Task 4 (issue #53): internal/host/tunnel_client.go dials CARLOS_TUNNEL_URL,
performs the /register handshake (Bearer token + Carlos-Tunnel-Name, waits
for 101 before writing anything else onto the conn), then inverts the
connection and serves the host's own edge.Edge over HTTP/2. Reconnects
forever with 250ms-floor/2s-cap jittered backoff on dial failure; a
successful dial that drops immediately re-dials without delay (reconnect
latency is user-visible per hq dual-WAN and Pi Wi-Fi blips). tunnelState()
exposes ""/"connected"/"reconnecting" for Task 6's heartbeat probe.

RunAgent now uses edge.RunWithActivatorReady's ready hook to start
tunnelLoop(e) on its own goroutine once the Edge is fully configured,
since the hook runs synchronously ahead of the blocking serve.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 7e13500a64a057bdf769a5386fd633883c8d3927 parent 3850c83
3 files changed, +227 −1
  • internal/host/agent.go +6 −1
  • internal/host/tunnel_client.go +147 −0
  • internal/host/tunnel_client_test.go +74 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 49d4d016d..59d4a5e07 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -154,7 +154,12 @@ func RunAgent(args []string) {
// The same serving loop as `carlos edge`, with the activator injected.
// Caching applies here too: cached hits still avoid the activator, which
// is the entire point of checking the cache before the hibernation branch.
- edge.RunWithActivator(args, act, cache)
+ // The ready hook fires synchronously, just before the blocking serve, so
+ // tunnelLoop is handed off to its own goroutine rather than called here —
+ // calling it directly would block the front door forever.
+ edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
+ go tunnelLoop(e)
+ })
// The edge has drained; take the children down too (they're this process's
// children — orphaning them would strand their leases until TTL expiry).
diff --git a/internal/host/tunnel_client.go b/internal/host/tunnel_client.go
new file mode 100644
index 000000000..fced7fbbf
--- /dev/null
+++ b/internal/host/tunnel_client.go
@@ -0,0 +1,147 @@
+// 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.
+package host
+
+import (
+ "bufio"
+ "crypto/tls"
+ "fmt"
+ "io"
+ "log"
+ "math/rand/v2"
+ "net"
+ "net/http"
+ "net/url"
+ "os"
+ "sync/atomic"
+ "time"
+
+ "golang.org/x/net/http2"
+)
+
+// tunnelStateVal is "" (disabled/never started), "connected", or
+// "reconnecting" — read by statusLoop's heartbeat probe (probes must fail
+// toward honest values; "" means the loop isn't running at all).
+var tunnelStateVal atomic.Value
+
+func tunnelState() string {
+ if v, ok := tunnelStateVal.Load().(string); ok {
+ return v
+ }
+ return ""
+}
+
+const (
+ tunnelBackoffFloor = 250 * time.Millisecond
+ tunnelBackoffCap = 2 * 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.
+func tunnelLoop(handler http.Handler) {
+ rawURL := os.Getenv("CARLOS_TUNNEL_URL")
+ token := os.Getenv("CARLOS_TUNNEL_TOKEN")
+ name := os.Getenv("CARLOS_TUNNEL_NAME")
+ if rawURL == "" || token == "" || name == "" {
+ if rawURL != "" || token != "" || name != "" {
+ log.Printf("agent: tunnel loop disabled: need all of CARLOS_TUNNEL_URL, CARLOS_TUNNEL_TOKEN, CARLOS_TUNNEL_NAME")
+ }
+ return
+ }
+ log.Printf("agent: tunnel loop armed (%s as %q)", rawURL, name)
+ backoff := tunnelBackoffFloor
+ for {
+ conn, err := dialTunnel(rawURL, name, token)
+ if err != nil {
+ tunnelStateVal.Store("reconnecting")
+ log.Printf("agent: tunnel dial: %v (retry in %s)", err, backoff)
+ time.Sleep(backoff + rand.N(backoff/2))
+ if backoff *= 2; backoff > tunnelBackoffCap {
+ backoff = tunnelBackoffCap
+ }
+ continue
+ }
+ backoff = tunnelBackoffFloor
+ tunnelStateVal.Store("connected")
+ log.Printf("agent: tunnel connected as %q", name)
+ serveTunnel(conn, handler) // blocks until the conn dies
+ tunnelStateVal.Store("reconnecting")
+ log.Printf("agent: tunnel lost — re-dialing")
+ }
+}
+
+// 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
+ var conn net.Conn
+ switch u.Scheme {
+ case "https":
+ if u.Port() == "" {
+ addr = net.JoinHostPort(u.Hostname(), "443")
+ }
+ conn, err = tls.Dial("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 = net.Dial("tcp", addr)
+ default:
+ return nil, fmt.Errorf("tunnel url: unsupported scheme %q", u.Scheme)
+ }
+ if err != nil {
+ return nil, 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 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})
+}
diff --git a/internal/host/tunnel_client_test.go b/internal/host/tunnel_client_test.go
new file mode 100644
index 000000000..035173b59
--- /dev/null
+++ b/internal/host/tunnel_client_test.go
@@ -0,0 +1,74 @@
+package host
+
+import (
+ "fmt"
+ "io"
+ "net/http"
+ "net/http/httptest"
+ "testing"
+
+ "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")
+ }
+}