feat(adapt): a net/http handler takes an upgrade's connection, so a WebSocket server behind AdaptNetHTTP answers 101, not 501

Claude-Session: https://claude.ai/code/session_01AGuBEqCjuman4E9FdKEgeV
This commit is contained in:
2026-10-03 00:03:26 -07:00
parent 863682e102
commit 0075bc119d
3 changed files with 148 additions and 0 deletions
+1
View File
@@ -56,6 +56,7 @@ Patch tags, one above the highest `v*` on GitHub. Push the commit to main, read
- The call-plane body is ZAP, laid out from struct field order (`internal/zapenc`). Append fields at the end. Reordering, inserting or retyping a field changes the wire.
- Doc prose is filed under `DocKey(pkg, method, path)`. zipdoc writes the package's import path. `pkgOf` reads the handler's package from its runtime name and maps `main` to the build info path, because a built binary names package main's functions `main.` while `go test` names them by import path.
- MCP over HTTP: POST only (GET and DELETE answer 405), one message per request, no sessions, no batches, notifications 202. `negotiate` answers `initialize` in the requested revision when it is 2025-06-18, 2025-11-25 or 2026-07-28, else in 2026-07-28. The official TypeScript (1.30.0) and Python (2.2.0) clients refuse an answer in a revision they did not request and do not know.
- `AdaptNetHTTP` carries an upgrade: a request with `Connection: Upgrade` runs the handler inside fasthttp's hijack, on a writer that is an `http.Hijacker`, so a net/http WebSocket server (coder/websocket, gorilla) writes its own 101 and owns the socket; a handler that declines answers an ordinary HTTP/1.1 reply. `wsx` stays the native way for a zip handler.
- zip validates no tokens. Identity is read from gateway-set headers (`HeaderOrg` is `X-Org-Id`); `Authorize` runs at the invoke seam, so it covers REST, MCP, the call plane and the in-process CLI.
- `app.Test` runs `prepare`, so `/mcp`, the OpenAPI document and the call plane answer under test as they do when serving.
- The `/docs` page loads Swagger UI from cdn.jsdelivr.net. Embedding swagger-ui-dist 5 would add about 1.74 MB (bundle 1,552,209 bytes, CSS 185,784 bytes, Apache-2.0) to every binary that links zip.
+87
View File
@@ -1,7 +1,12 @@
package zip
import (
"bufio"
"bytes"
"context"
"fmt"
"io"
"net"
"net/http"
"strconv"
"sync"
@@ -98,6 +103,16 @@ func adaptStreaming(h http.Handler) func(*Ctx) error {
// matches on URL.Path, which is why a mux in front of it looks fine.
req.RequestURI = string(c.fc.Request().Header.RequestURI())
// AN UPGRADE IS ANSWERED ON THE CONNECTION ITSELF. A net/http WebSocket
// server (coder/websocket, gorilla) takes the socket with http.Hijacker and
// writes the 101 itself; a writer without Hijack makes it answer 501 to
// every upgrade. So the handler runs inside fasthttp's hijack, against the
// raw connection, after this returns.
if upgrading(c.fc.Request()) {
serveUpgrade(c, h, req)
return nil
}
pr, pw := io.Pipe()
w := &streamWriter{hdr: make(http.Header), status: http.StatusOK, pw: pw, ready: make(chan struct{})}
@@ -136,3 +151,75 @@ func adaptStreaming(h http.Handler) func(*Ctx) error {
return nil
}
}
// serveUpgrade runs h on the hijacked connection. The request no longer has the
// fasthttp context to carry — the handler must not hold ctx members past the
// hijack — so it carries one of its own, cancelled when the handler returns.
func serveUpgrade(c *Ctx, h http.Handler, req *http.Request) {
ctx, cancel := context.WithCancel(context.Background())
req = req.WithContext(ctx)
rc := c.fc.RequestCtx()
rc.HijackSetNoResponse(true)
rc.Hijack(func(conn net.Conn) {
defer cancel()
w := &upgradeWriter{
conn: conn,
brw: bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)),
hdr: make(http.Header),
}
h.ServeHTTP(w, req)
if !w.taken {
w.answer()
}
})
}
// upgradeWriter is the ResponseWriter of an upgrade request. A handler that
// takes the connection (Hijack) speaks the new protocol on it; one that declines
// — an unauthorized socket is refused before it opens — has its status, headers
// and body written as an ordinary HTTP/1.1 reply, and the connection closes.
type upgradeWriter struct {
conn net.Conn
brw *bufio.ReadWriter
hdr http.Header
status int
body bytes.Buffer
taken bool
}
func (w *upgradeWriter) Header() http.Header { return w.hdr }
func (w *upgradeWriter) WriteHeader(code int) {
if w.status == 0 {
w.status = code
}
}
func (w *upgradeWriter) Write(b []byte) (int, error) {
w.WriteHeader(http.StatusOK)
return w.body.Write(b)
}
// Flush satisfies http.Flusher; nothing is sent until the handler returns or
// takes the connection.
func (w *upgradeWriter) Flush() {}
// Hijack hands the handler the connection, and with it the whole reply.
func (w *upgradeWriter) Hijack() (net.Conn, *bufio.ReadWriter, error) {
if w.taken {
return nil, nil, http.ErrHijacked
}
w.taken = true
return w.conn, w.brw, nil
}
func (w *upgradeWriter) answer() {
w.WriteHeader(http.StatusOK)
fmt.Fprintf(w.brw, "HTTP/1.1 %d %s\r\n", w.status, http.StatusText(w.status))
w.hdr.Set("Content-Length", strconv.Itoa(w.body.Len()))
w.hdr.Set("Connection", "close")
_ = w.hdr.Write(w.brw)
_, _ = w.brw.WriteString("\r\n")
_, _ = w.brw.Write(w.body.Bytes())
_ = w.brw.Flush()
}
+60
View File
@@ -144,3 +144,63 @@ func TestAdaptNetHTTP_PanicClosesTheStream(t *testing.T) {
t.Fatal("reader stranded after the handler panicked")
}
}
// AN UPGRADE REACHES THE CONNECTION. A net/http WebSocket server takes the
// socket with http.Hijacker and writes the 101 itself; without Hijack on the
// writer coder/websocket answers 501 to every upgrade, which is how the talk
// socket behind api.hanzo.ai refused every browser that reached it.
func TestAdaptNetHTTP_AnUpgradeTakesTheConnection(t *testing.T) {
base := streamApp(t, func(w http.ResponseWriter, r *http.Request) {
if r.URL.Query().Get("ticket") != "ok" {
http.Error(w, "a valid ticket is required", http.StatusUnauthorized)
return
}
conn, brw, err := w.(http.Hijacker).Hijack()
if err != nil {
t.Errorf("hijack: %v", err)
return
}
defer conn.Close()
_, _ = brw.WriteString("HTTP/1.1 101 Switching Protocols\r\nUpgrade: echo\r\nConnection: Upgrade\r\n\r\n")
_ = brw.Flush()
line, _ := brw.ReadString('\n')
_, _ = brw.WriteString("echo " + line)
_ = brw.Flush()
})
addr := strings.TrimPrefix(base, "http://")
dial := func(ticket string) (net.Conn, *bufio.Reader) {
t.Helper()
conn, err := net.Dial("tcp", addr)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { _ = conn.Close() })
_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
fmt.Fprintf(conn, "GET /legacy/socket?ticket=%s HTTP/1.1\r\nHost: x\r\nConnection: Upgrade\r\nUpgrade: echo\r\n\r\n", ticket)
return conn, bufio.NewReader(conn)
}
conn, rd := dial("ok")
status, _ := rd.ReadString('\n')
if !strings.HasPrefix(status, "HTTP/1.1 101") {
t.Fatalf("upgrade answered %q, want 101 from the handler", status)
}
for line, _ := rd.ReadString('\n'); line != "\r\n" && line != ""; line, _ = rd.ReadString('\n') {
}
fmt.Fprint(conn, "ping\n")
if got, _ := rd.ReadString('\n'); got != "echo ping\n" {
t.Fatalf("the upgraded connection answered %q, want %q", got, "echo ping\n")
}
// A handler that declines the upgrade answers an ordinary reply.
_, rd = dial("no")
status, _ = rd.ReadString('\n')
if !strings.HasPrefix(status, "HTTP/1.1 401") {
t.Fatalf("a refused upgrade answered %q, want 401", status)
}
rest, _ := io.ReadAll(rd)
if !strings.Contains(string(rest), "a valid ticket is required") {
t.Fatalf("the refusal lost its body: %q", rest)
}
}