Skip to content

Commit f4b88b9

Browse files
lemon-mintlemon-mint
committed
feat(http1): added HTTP/1.1 idle connection pooling with reuse
- Implement HTTP/1.1 connection pool with 90 second idle timeout and reuse. - Close connections on body cancellation or Connection: close header. - Add connection lifecycle methods: adoptH1, releaseH1, closeH1 with pooling. - Add HTTP/1.1 keep-alive tests: EOF reuse, partial body rejection, close header. - Add HTTP/2 roundtrip tests: client connection usage and pooling validation. - Update ARCHITECTURE.md to document HTTP/1.1 idle connection pooling behavior.
1 parent a460be5 commit f4b88b9

4 files changed

Lines changed: 464 additions & 19 deletions

File tree

‎ARCHITECTURE.md‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,7 @@ The Service Worker keeps tab, entry, client-context, route, and stream maps in m
9797

9898
The Go WASM kernel exposes `__zp_kernel_init`, `__go_jshttp`, `__zp_stream`, and `__zp_cookie_set` to the Service Worker. Initialization creates a browser WebSocket to `/__zp/ws-pipe`; the relay accepts it with Gorilla WebSocket and adapts binary messages to a stream-oriented `net.Conn`. A yamux client/server session runs over that connection, and per-target yamux streams are bridged by the relay to Tor SOCKS5.
9999

100-
`internal/http1` builds sanitized target requests, applies the cookie jar, follows redirects up to `MaxRedirects`, dispatches HTTPS fetches to HTTP/2 when ALPN selects `h2`, and falls back to direct HTTP/1.1 request/response handling otherwise. HTTP/2 client connections are pooled only within the same target authority, tab, and Tor isolation token; target WebSocket support stays HTTP/1.1 Upgrade through `internal/wsproto` and the runtime `WebSocket` wrapper.
100+
`internal/http1` builds sanitized target requests, applies the cookie jar, follows redirects up to `MaxRedirects`, dispatches HTTPS fetches to HTTP/2 when ALPN selects `h2`, and falls back to direct HTTP/1.1 request/response handling otherwise. HTTP/2 client connections and reusable HTTP/1.1 idle connections are pooled only within the same target authority, tab, and Tor isolation token. Idle target connections use a browser-style 90 second timeout; HTTP/1.1 connections are reused only after the response body reaches EOF and are closed on partial body cancellation or `Connection: close`. Target WebSocket support stays HTTP/1.1 Upgrade through `internal/wsproto` and the runtime `WebSocket` wrapper.
101101

102102
`internal/swhttp.ResponseToJS` constructs JavaScript `Response` objects with a `ReadableStream` backed by the Go response body. Document HTML transformation uses `htmltx.TransformTo` through an `io.Pipe`, so transformed HTML can flow to the browser without first buffering the full document. Request/upload body conversion and browser backpressure/cancellation fidelity are still prototype-level.
103103

‎internal/http1/http2_test.go‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,76 @@ func TestHTTP2RoundTripUsesClientConn(t *testing.T) {
7373
}
7474
}
7575

76+
func TestHTTP2KeepAliveReusesPooledClientConn(t *testing.T) {
77+
client, server := net.Pipe()
78+
paths := make(chan string, 2)
79+
go func() {
80+
defer server.Close()
81+
(&http2.Server{}).ServeConn(server, &http2.ServeConnOpts{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
82+
paths <- r.URL.RequestURI()
83+
_, _ = io.WriteString(w, "h2-ok")
84+
})})
85+
}()
86+
87+
engine := &Engine{}
88+
tab := &TabState{CookieJar: cookiejar.New(), StreamIsolationKey: []byte("0123456789abcdef0123456789abcdef")}
89+
first, _ := url.Parse("https://example.com/first")
90+
hc, err := newH2Conn(client)
91+
if err != nil {
92+
t.Fatal(err)
93+
}
94+
hc = engine.adoptH2(h2PoolKey(first, tab), hc)
95+
t.Cleanup(hc.close)
96+
97+
wireReq, err := BuildHTTP1Request(mustRequest(t, first), first, tab.CookieJar)
98+
if err != nil {
99+
t.Fatal(err)
100+
}
101+
resp, err := engine.roundTripHTTP2(context.Background(), hc, wireReq)
102+
if err != nil {
103+
t.Fatal(err)
104+
}
105+
readAndCloseH2(t, resp)
106+
107+
second := *first
108+
second.Path = "/second"
109+
resp, err = engine.RoundTrip(context.Background(), mustRequest(t, &second), &second, tab)
110+
if err != nil {
111+
t.Fatal(err)
112+
}
113+
readAndCloseH2(t, resp)
114+
115+
if got := <-paths; got != "/first" {
116+
t.Fatalf("first path = %q", got)
117+
}
118+
if got := <-paths; got != "/second" {
119+
t.Fatalf("second path = %q", got)
120+
}
121+
}
122+
123+
func mustRequest(t *testing.T, target *url.URL) *http.Request {
124+
t.Helper()
125+
req, err := http.NewRequest(http.MethodGet, target.String(), nil)
126+
if err != nil {
127+
t.Fatal(err)
128+
}
129+
return req
130+
}
131+
132+
func readAndCloseH2(t *testing.T, resp *http.Response) {
133+
t.Helper()
134+
body, err := io.ReadAll(resp.Body)
135+
if err != nil {
136+
t.Fatal(err)
137+
}
138+
if string(body) != "h2-ok" {
139+
t.Fatalf("body = %q", string(body))
140+
}
141+
if err := resp.Body.Close(); err != nil {
142+
t.Fatal(err)
143+
}
144+
}
145+
76146
func validateHTTP2Request(r *http.Request) error {
77147
if r.ProtoMajor != 2 {
78148
return fmt.Errorf("proto = %s", r.Proto)

‎internal/http1/keepalive_test.go‎

Lines changed: 173 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,173 @@
1+
package http1
2+
3+
import (
4+
"bufio"
5+
"context"
6+
"fmt"
7+
"io"
8+
"net"
9+
"net/http"
10+
"net/url"
11+
"testing"
12+
"time"
13+
14+
"github.com/gosuda/zeroproxy/internal/cookiejar"
15+
)
16+
17+
func TestHTTP1KeepAliveReusesIdleConnectionAfterEOF(t *testing.T) {
18+
mux := &pipeMux{streams: make(chan net.Conn, 2)}
19+
engine := &Engine{Mux: mux}
20+
tab := &TabState{CookieJar: cookiejar.New(), StreamIsolationKey: []byte("0123456789abcdef0123456789abcdef")}
21+
target, _ := url.Parse("http://example.com/one")
22+
serverDone := make(chan error, 1)
23+
go func() {
24+
c := <-mux.streams
25+
defer c.Close()
26+
br := bufio.NewReader(c)
27+
if err := acceptSOCKSConnect(br, c); err != nil {
28+
serverDone <- err
29+
return
30+
}
31+
for _, wantPath := range []string{"/one", "/two"} {
32+
req, err := http.ReadRequest(br)
33+
if err != nil {
34+
serverDone <- err
35+
return
36+
}
37+
if req.URL.RequestURI() != wantPath {
38+
serverDone <- fmt.Errorf("request path = %q, want %q", req.URL.RequestURI(), wantPath)
39+
return
40+
}
41+
if _, err := io.WriteString(c, "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok"); err != nil {
42+
serverDone <- err
43+
return
44+
}
45+
}
46+
serverDone <- nil
47+
}()
48+
49+
roundTripAndClose(t, engine, tab, target)
50+
second := *target
51+
second.Path = "/two"
52+
roundTripAndClose(t, engine, tab, &second)
53+
if err := <-serverDone; err != nil {
54+
t.Fatal(err)
55+
}
56+
select {
57+
case c := <-mux.streams:
58+
_ = c.Close()
59+
t.Fatal("HTTP/1.1 keep-alive opened a second target connection")
60+
default:
61+
}
62+
}
63+
64+
func TestHTTP1DoesNotReuseWhenBodyClosedBeforeEOF(t *testing.T) {
65+
mux := &pipeMux{streams: make(chan net.Conn, 2)}
66+
engine := &Engine{Mux: mux}
67+
tab := &TabState{CookieJar: cookiejar.New(), StreamIsolationKey: []byte("0123456789abcdef0123456789abcdef")}
68+
target, _ := url.Parse("http://example.com/partial")
69+
firstDone := serveOneHTTP1Response(t, mux, "/partial", "HTTP/1.1 200 OK\r\nContent-Length: 4\r\n\r\nbody")
70+
71+
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
72+
defer cancel()
73+
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil)
74+
resp, err := engine.RoundTrip(ctx, req, target, tab)
75+
if err != nil {
76+
t.Fatal(err)
77+
}
78+
buf := make([]byte, 1)
79+
if _, err := io.ReadFull(resp.Body, buf); err != nil {
80+
t.Fatal(err)
81+
}
82+
if err := resp.Body.Close(); err != nil {
83+
t.Fatal(err)
84+
}
85+
assertNoIdleHTTP1(t, engine, target, tab)
86+
if err := <-firstDone; err != nil {
87+
t.Fatal(err)
88+
}
89+
90+
second := *target
91+
second.Path = "/after-partial-close"
92+
secondDone := serveOneHTTP1Response(t, mux, "/after-partial-close", "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
93+
roundTripAndClose(t, engine, tab, &second)
94+
if err := <-secondDone; err != nil {
95+
t.Fatal(err)
96+
}
97+
}
98+
99+
func TestHTTP1DoesNotReuseConnectionCloseResponse(t *testing.T) {
100+
mux := &pipeMux{streams: make(chan net.Conn, 2)}
101+
engine := &Engine{Mux: mux}
102+
tab := &TabState{CookieJar: cookiejar.New(), StreamIsolationKey: []byte("0123456789abcdef0123456789abcdef")}
103+
target, _ := url.Parse("http://example.com/close")
104+
firstDone := serveOneHTTP1Response(t, mux, "/close", "HTTP/1.1 200 OK\r\nConnection: close\r\nContent-Length: 2\r\n\r\nok")
105+
roundTripAndClose(t, engine, tab, target)
106+
assertNoIdleHTTP1(t, engine, target, tab)
107+
if err := <-firstDone; err != nil {
108+
t.Fatal(err)
109+
}
110+
111+
second := *target
112+
second.Path = "/after-close"
113+
secondDone := serveOneHTTP1Response(t, mux, "/after-close", "HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok")
114+
roundTripAndClose(t, engine, tab, &second)
115+
if err := <-secondDone; err != nil {
116+
t.Fatal(err)
117+
}
118+
}
119+
120+
func roundTripAndClose(t *testing.T, engine *Engine, tab *TabState, target *url.URL) {
121+
t.Helper()
122+
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
123+
defer cancel()
124+
req, _ := http.NewRequestWithContext(ctx, http.MethodGet, target.String(), nil)
125+
resp, err := engine.RoundTrip(ctx, req, target, tab)
126+
if err != nil {
127+
t.Fatal(err)
128+
}
129+
body, err := io.ReadAll(resp.Body)
130+
if err != nil {
131+
t.Fatal(err)
132+
}
133+
if string(body) != "ok" {
134+
t.Fatalf("body = %q", string(body))
135+
}
136+
if err := resp.Body.Close(); err != nil {
137+
t.Fatal(err)
138+
}
139+
}
140+
141+
func assertNoIdleHTTP1(t *testing.T, engine *Engine, target *url.URL, tab *TabState) {
142+
t.Helper()
143+
if pc := engine.reserveH1(h2PoolKey(target, tab)); pc != nil {
144+
engine.closeH1(pc)
145+
t.Fatal("HTTP/1.1 connection was left idle when it should have been closed")
146+
}
147+
}
148+
149+
func serveOneHTTP1Response(t *testing.T, mux *pipeMux, wantPath, response string) <-chan error {
150+
t.Helper()
151+
done := make(chan error, 1)
152+
go func() {
153+
c := <-mux.streams
154+
defer c.Close()
155+
br := bufio.NewReader(c)
156+
if err := acceptSOCKSConnect(br, c); err != nil {
157+
done <- err
158+
return
159+
}
160+
req, err := http.ReadRequest(br)
161+
if err != nil {
162+
done <- err
163+
return
164+
}
165+
if req.URL.RequestURI() != wantPath {
166+
done <- fmt.Errorf("request path = %q, want %q", req.URL.RequestURI(), wantPath)
167+
return
168+
}
169+
_, err = io.WriteString(c, response)
170+
done <- err
171+
}()
172+
return done
173+
}

0 commit comments

Comments
 (0)