From 8efcdc486dadd44fc0594886f64414ad39bd2a39 Mon Sep 17 00:00:00 2001 From: Adam Fisk Date: Thu, 22 Oct 2026 09:44:13 -0600 Subject: [PATCH 1/2] roundtripper: give back the stream an unused round tripper holds NewRoundTripper confirms its connection before returning, and that connection holds one of the session's streams until a request takes it. kindling discards connected round trippers that lose a race, so on a client that keeps losing short races (Unbounded's signaling polls) the streams leak. The session then never looks idle, so it's never retired, and once MaxStreams have leaked every dial fails with "soar: too many streams" until the app restarts. One Windows beta client did that ~1,800 times in two days. The returned transport now implements io.Closer, which releases an untaken connection, and an untaken connection closes itself after preDialedTTL for callers that never close it. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QQFpfCXs3JqrfyDuiGabF5 --- e2e_test.go | 55 ++++++++++++++++++++++++++++++++++++ roundtripper.go | 75 ++++++++++++++++++++++++++++++++++++++++--------- 2 files changed, 116 insertions(+), 14 deletions(-) diff --git a/e2e_test.go b/e2e_test.go index c34dc42..be43511 100644 --- a/e2e_test.go +++ b/e2e_test.go @@ -1093,6 +1093,61 @@ func TestNewRoundTripperFailsWhenDestinationUnreachable(t *testing.T) { } } +// An unused round tripper holds a stream until it's closed. A race that discards its losers +// must get those streams back, or the session runs out of them for good. +func TestClosedRoundTrippersGiveBackTheirStreams(t *testing.T) { + srv, pub := startServer(t) + hs := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + t.Cleanup(hs.Close) + addr := strings.TrimPrefix(hs.URL, "http://") + c, _ := NewClient(Config{Zone: testZone, ServerKey: pub, Resolvers: []string{srv}, MaxStreams: 2}) + defer c.Close() + for range 2 { + rt, err := c.NewRoundTripper(context.Background(), addr) + if err != nil { + t.Fatal(err) + } + rt.(io.Closer).Close() + } + if _, err := connectWithin(c, addr, 10*time.Second); err != nil { + t.Fatalf("connecting after closing every unused round tripper: %v", err) + } +} + +func TestUnusedRoundTripperGivesBackItsStream(t *testing.T) { + defer func(ttl time.Duration) { preDialedTTL = ttl }(preDialedTTL) + preDialedTTL = 200 * time.Millisecond + srv, pub := startServer(t) + hs := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) + t.Cleanup(hs.Close) + addr := strings.TrimPrefix(hs.URL, "http://") + c, _ := NewClient(Config{Zone: testZone, ServerKey: pub, Resolvers: []string{srv}, MaxStreams: 2}) + defer c.Close() + for range 2 { + if _, err := c.NewRoundTripper(context.Background(), addr); err != nil { + t.Fatal(err) + } + } + if _, err := connectWithin(c, addr, 10*time.Second); err != nil { + t.Fatalf("connecting after the unused round trippers expired: %v", err) + } +} + +// connectWithin retries NewRoundTripper until it succeeds: a closed stream gives back its slot +// only once the server acknowledges its FIN. +func connectWithin(c *Client, addr string, d time.Duration) (http.RoundTripper, error) { + deadline := time.Now().Add(d) + for { + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + rt, err := c.NewRoundTripper(ctx, addr) + cancel() + if err == nil || time.Now().After(deadline) { + return rt, err + } + time.Sleep(100 * time.Millisecond) + } +} + // errInjector forwards to server but first answers every query with a forged error. func errInjector(t *testing.T, server string, rcode byte, tc bool) string { front, _ := net.ListenPacket("udp", "127.0.0.1:0") diff --git a/roundtripper.go b/roundtripper.go index a507389..d06fe85 100644 --- a/roundtripper.go +++ b/roundtripper.go @@ -14,6 +14,10 @@ import ( // phase and falls back to another transport only when connecting fails. TLS runs end to end, so // the server only relays ciphertext. With Close, MaxLength and RequestTimeout, this satisfies // kindling's DNS-tunnel transport (dnstt.DNSTT). +// +// The confirmed connection holds one of the session's streams until a request takes it. Close +// the returned transport (it implements io.Closer) when it won't be used, as a race does when +// another transport wins; otherwise the connection is closed after preDialedTTL. func (c *Client) NewRoundTripper(ctx context.Context, addr string) (http.RoundTripper, error) { var pre *preDialed if addr != "" { @@ -26,29 +30,54 @@ func (c *Client) NewRoundTripper(ctx context.Context, addr string) (http.RoundTr return nil, err } pre = &preDialed{conn: conn, addr: addr} + pre.mu.Lock() + pre.timer = time.AfterFunc(preDialedTTL, pre.close) + pre.mu.Unlock() } else if _, err := c.session(ctx); err != nil { return nil, err } - return &http.Transport{ - DialContext: func(ctx context.Context, network, a string) (net.Conn, error) { - if conn := pre.take(a); conn != nil { - return conn, nil - } - return c.DialContext(ctx, network, a) + return &roundTripper{ + Transport: &http.Transport{ + DialContext: func(ctx context.Context, network, a string) (net.Conn, error) { + if conn := pre.take(a); conn != nil { + return conn, nil + } + return c.DialContext(ctx, network, a) + }, + MaxIdleConns: 4, + MaxIdleConnsPerHost: 2, + // Idle keep-alives hold a stream open, and an open stream keeps a poll outstanding. + IdleConnTimeout: 5 * time.Second, + TLSHandshakeTimeout: 30 * time.Second, }, - MaxIdleConns: 4, - MaxIdleConnsPerHost: 2, - // Idle keep-alives hold a stream open, and an open stream keeps a poll outstanding. - IdleConnTimeout: 5 * time.Second, - TLSHandshakeTimeout: 30 * time.Second, + pre: pre, }, nil } +// preDialedTTL bounds how long a confirmed connection waits for its first request. A session +// has only MaxStreams streams, so one never taken must not hold its stream for good. +var preDialedTTL = 10 * time.Second + +// roundTripper is the transport NewRoundTripper returns. +type roundTripper struct { + *http.Transport + pre *preDialed +} + +// Close releases the confirmed connection if no request took it, and closes idle connections. +// Requests already in flight are unaffected. +func (rt *roundTripper) Close() error { + rt.pre.close() + rt.CloseIdleConnections() + return nil +} + // preDialed hands a confirmed connection to the first dial for its address. type preDialed struct { - mu sync.Mutex - conn net.Conn - addr string + mu sync.Mutex + conn net.Conn + addr string + timer *time.Timer } func (p *preDialed) take(addr string) net.Conn { @@ -62,9 +91,27 @@ func (p *preDialed) take(addr string) net.Conn { } conn := p.conn p.conn = nil + p.timer.Stop() return conn } +// close closes the connection unless a dial already took it. +func (p *preDialed) close() { + if p == nil { + return + } + p.mu.Lock() + conn := p.conn + p.conn = nil + if p.timer != nil { + p.timer.Stop() + } + p.mu.Unlock() + if conn != nil { + conn.Close() + } +} + // MaxLength caps request bodies: uploads ride ~50-byte queries, so big ones belong elsewhere. func (c *Client) MaxLength() int { return 64 << 10 } From c5d026ff2aac281c4d2d4999e26881e94f7a9a10 Mon Sep 17 00:00:00 2001 From: Adam Fisk Date: Thu, 22 Oct 2026 09:56:35 -0600 Subject: [PATCH 2/2] e2e_test: keep expiry out of the explicit-close test Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01QQFpfCXs3JqrfyDuiGabF5 --- e2e_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/e2e_test.go b/e2e_test.go index be43511..f721605 100644 --- a/e2e_test.go +++ b/e2e_test.go @@ -1096,6 +1096,9 @@ func TestNewRoundTripperFailsWhenDestinationUnreachable(t *testing.T) { // An unused round tripper holds a stream until it's closed. A race that discards its losers // must get those streams back, or the session runs out of them for good. func TestClosedRoundTrippersGiveBackTheirStreams(t *testing.T) { + // Only Close may give the streams back here, not their expiry. + defer func(ttl time.Duration) { preDialedTTL = ttl }(preDialedTTL) + preDialedTTL = time.Hour srv, pub := startServer(t) hs := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {})) t.Cleanup(hs.Close)