From 0b80a6919212ec2b6c2a5a6fa68535423ffd697c Mon Sep 17 00:00:00 2001 From: penguinehis Date: Fri, 10 Jul 2026 11:52:08 -0300 Subject: [PATCH] Fix memory leak --- dnstt_integration.go | 11 +- internal/turbotunnel/clientid.go | 37 ++++ internal/turbotunnel/consts.go | 22 +++ internal/turbotunnel/queuepacketconn.go | 165 ++++++++++++++++ internal/turbotunnel/remotemap.go | 198 ++++++++++++++++++++ internal/turbotunnel/remotemap_leak_test.go | 51 +++++ xray_integration.go | 22 +++ xray_native_tuning.go | 11 ++ xray_xhttp.go | 98 +++++++++- 9 files changed, 608 insertions(+), 7 deletions(-) create mode 100644 internal/turbotunnel/clientid.go create mode 100644 internal/turbotunnel/consts.go create mode 100644 internal/turbotunnel/queuepacketconn.go create mode 100644 internal/turbotunnel/remotemap.go create mode 100644 internal/turbotunnel/remotemap_leak_test.go diff --git a/dnstt_integration.go b/dnstt_integration.go index 612b735..07e6f05 100644 --- a/dnstt_integration.go +++ b/dnstt_integration.go @@ -34,7 +34,12 @@ import ( "golang.org/x/crypto/ssh" "www.bamsoftware.com/git/dnstt.git/dns" "www.bamsoftware.com/git/dnstt.git/noise" - "www.bamsoftware.com/git/dnstt.git/turbotunnel" + + // Local fork of the upstream turbotunnel package. Identical except that + // RemoteMap's background expiry goroutine is stoppable via Close(); upstream + // leaks it for the process lifetime, which becomes a per-restart goroutine + // leak under this integration's DNSTT hot-reload / auto-restart. + "shell2/internal/turbotunnel" ) // ---------- Hot-reload stop mechanism ---------- @@ -1664,6 +1669,10 @@ func runDNSTTOnListeners(privkey []byte, listeners []dnsttListenerSpec, limits d } // set up turbotunnel and KCP listener ttConn := turbotunnel.NewQueuePacketConn(turbotunnel.DummyAddr{}, idleTimeout*2) + // Close ttConn on exit so its RemoteMap expiry goroutine is stopped. Without + // this, every DNSTT restart (hot-reload / auto-restart) leaks that goroutine + // plus the QueuePacketConn's buffers for the process lifetime. + defer ttConn.Close() ln, err := kcp.ServeConn(nil, 0, 0, ttConn) if err != nil { return fmt.Errorf("dnstt: opening KCP listener: %v", err) diff --git a/internal/turbotunnel/clientid.go b/internal/turbotunnel/clientid.go new file mode 100644 index 0000000..06ff837 --- /dev/null +++ b/internal/turbotunnel/clientid.go @@ -0,0 +1,37 @@ +// This package is a local fork of www.bamsoftware.com/git/dnstt.git/turbotunnel +// (upstream v1.20241021.0). The only behavioural change from upstream is that +// RemoteMap's background expiry goroutine is now stoppable via Close(), wired +// through QueuePacketConn.Close(); see remotemap.go and queuepacketconn.go. +// Upstream leaks that goroutine for the process lifetime, which is harmless for +// the upstream one-shot server but leaks one goroutine per DNSTT restart in this +// integration (hot-reload / auto-restart). clientid.go and consts.go are copied +// verbatim. + +package turbotunnel + +import ( + "crypto/rand" + "encoding/hex" +) + +// ClientID is an abstract identifier that binds together all the communications +// belonging to a single client session, even though those communications may +// arrive from multiple IP addresses or over multiple lower-level connections. +// It plays the same role that an (IP address, port number) tuple plays in a +// net.UDPConn: it's the return address pertaining to a long-lived abstract +// client session. The client attaches its ClientID to each of its +// communications, enabling the server to disambiguate requests among its many +// clients. ClientID implements the net.Addr interface. +type ClientID [8]byte + +func NewClientID() ClientID { + var id ClientID + _, err := rand.Read(id[:]) + if err != nil { + panic(err) + } + return id +} + +func (id ClientID) Network() string { return "clientid" } +func (id ClientID) String() string { return hex.EncodeToString(id[:]) } diff --git a/internal/turbotunnel/consts.go b/internal/turbotunnel/consts.go new file mode 100644 index 0000000..5684bf7 --- /dev/null +++ b/internal/turbotunnel/consts.go @@ -0,0 +1,22 @@ +// Package turbotunnel is facilities for embedding packet-based reliability +// protocols inside other protocols. +// +// https://github.com/net4people/bbs/issues/9 +package turbotunnel + +import "errors" + +// QueueSize is the size of send and receive queues in QueuePacketConn and +// RemoteMap. +const QueueSize = 128 + +var errClosedPacketConn = errors.New("operation on closed connection") +var errNotImplemented = errors.New("not implemented") + +// DummyAddr is a placeholder net.Addr, for when a programming interface +// requires a net.Addr but there is none relevant. All DummyAddrs compare equal +// to each other. +type DummyAddr struct{} + +func (addr DummyAddr) Network() string { return "dummy" } +func (addr DummyAddr) String() string { return "dummy" } diff --git a/internal/turbotunnel/queuepacketconn.go b/internal/turbotunnel/queuepacketconn.go new file mode 100644 index 0000000..2097cca --- /dev/null +++ b/internal/turbotunnel/queuepacketconn.go @@ -0,0 +1,165 @@ +package turbotunnel + +import ( + "net" + "sync" + "sync/atomic" + "time" +) + +// taggedPacket is a combination of a []byte and a net.Addr, encapsulating the +// return type of PacketConn.ReadFrom. +type taggedPacket struct { + P []byte + Addr net.Addr +} + +// QueuePacketConn implements net.PacketConn by storing queues of packets. There +// is one incoming queue (where packets are additionally tagged by the source +// address of the peer that sent them). There are many outgoing queues, one for +// each remote peer address that has been recently seen. The QueueIncoming +// method inserts a packet into the incoming queue, to eventually be returned by +// ReadFrom. WriteTo inserts a packet into an address-specific outgoing queue, +// which can later by accessed through the OutgoingQueue method. +// +// Besides the outgoing queues, there is also a one-element "stash" for each +// remote peer address. You can stash a packet using the Stash method, and get +// it back later by receiving from the channel returned by Unstash. The stash is +// meant as a convenient place to temporarily store a single packet, such as +// when you've read one too many packets from the send queue and need to store +// the extra packet to be processed first in the next pass. It's the caller's +// responsibility to Unstash what they have Stashed. Calling Stash does not put +// the packet at the head of the send queue; if there is the possibility that a +// packet has been stashed, it must be checked for by calling Unstash in +// addition to OutgoingQueue. +type QueuePacketConn struct { + remotes *RemoteMap + localAddr net.Addr + recvQueue chan taggedPacket + closeOnce sync.Once + closed chan struct{} + // What error to return when the QueuePacketConn is closed. + err atomic.Value +} + +// NewQueuePacketConn makes a new QueuePacketConn, set to track recent peers +// for at least a duration of timeout. +func NewQueuePacketConn(localAddr net.Addr, timeout time.Duration) *QueuePacketConn { + return &QueuePacketConn{ + remotes: NewRemoteMap(timeout), + localAddr: localAddr, + recvQueue: make(chan taggedPacket, QueueSize), + closed: make(chan struct{}), + } +} + +// QueueIncoming queues and incoming packet and its source address, to be +// returned in a future call to ReadFrom. +func (c *QueuePacketConn) QueueIncoming(p []byte, addr net.Addr) { + select { + case <-c.closed: + // If we're closed, silently drop it. + return + default: + } + // Copy the slice so that the caller may reuse it. + buf := make([]byte, len(p)) + copy(buf, p) + select { + case c.recvQueue <- taggedPacket{buf, addr}: + default: + // Drop the incoming packet if the receive queue is full. + } +} + +// OutgoingQueue returns the queue of outgoing packets corresponding to addr, +// creating it if necessary. The contents of the queue will be packets that are +// written to the address in question using WriteTo. +func (c *QueuePacketConn) OutgoingQueue(addr net.Addr) <-chan []byte { + return c.remotes.SendQueue(addr) +} + +// Stash places p in the stash for addr, if the stash is not already occupied. +// Returns true if the packet was placed in the stash, or false if the stash was +// already occupied. This method is similar to WriteTo, except that it puts the +// packet in the stash queue (accessible via Unstash), rather than the outgoing +// queue (accessible via OutgoingQueue). +func (c *QueuePacketConn) Stash(p []byte, addr net.Addr) bool { + return c.remotes.Stash(addr, p) +} + +// Unstash returns the channel that represents the stash for addr. +func (c *QueuePacketConn) Unstash(addr net.Addr) <-chan []byte { + return c.remotes.Unstash(addr) +} + +// ReadFrom returns a packet and address previously stored by QueueIncoming. +func (c *QueuePacketConn) ReadFrom(p []byte) (int, net.Addr, error) { + select { + case <-c.closed: + return 0, nil, &net.OpError{Op: "read", Net: c.LocalAddr().Network(), Addr: c.LocalAddr(), Err: c.err.Load().(error)} + default: + } + select { + case <-c.closed: + return 0, nil, &net.OpError{Op: "read", Net: c.LocalAddr().Network(), Addr: c.LocalAddr(), Err: c.err.Load().(error)} + case packet := <-c.recvQueue: + return copy(p, packet.P), packet.Addr, nil + } +} + +// WriteTo queues an outgoing packet for the given address. The queue can later +// be retrieved using the OutgoingQueue method. +func (c *QueuePacketConn) WriteTo(p []byte, addr net.Addr) (int, error) { + select { + case <-c.closed: + return 0, &net.OpError{Op: "write", Net: c.LocalAddr().Network(), Addr: c.LocalAddr(), Err: c.err.Load().(error)} + default: + } + // Copy the slice so that the caller may reuse it. + buf := make([]byte, len(p)) + copy(buf, p) + select { + case c.remotes.SendQueue(addr) <- buf: + return len(buf), nil + default: + // Drop the outgoing packet if the send queue is full. + return len(buf), nil + } +} + +// closeWithError unblocks pending operations and makes future operations fail +// with the given error. If err is nil, it becomes errClosedPacketConn. +func (c *QueuePacketConn) closeWithError(err error) error { + var newlyClosed bool + c.closeOnce.Do(func() { + newlyClosed = true + // Store the error to be returned by future PacketConn + // operations. + if err == nil { + err = errClosedPacketConn + } + c.err.Store(err) + close(c.closed) + // LOCAL FORK ADDITION: stop the RemoteMap expiry goroutine so it is not + // leaked for the process lifetime. Upstream never does this. + c.remotes.Close() + }) + if !newlyClosed { + return &net.OpError{Op: "close", Net: c.LocalAddr().Network(), Addr: c.LocalAddr(), Err: c.err.Load().(error)} + } + return nil +} + +// Close unblocks pending operations and makes future operations fail with a +// "closed connection" error. +func (c *QueuePacketConn) Close() error { + return c.closeWithError(nil) +} + +// LocalAddr returns the localAddr value that was passed to NewQueuePacketConn. +func (c *QueuePacketConn) LocalAddr() net.Addr { return c.localAddr } + +func (c *QueuePacketConn) SetDeadline(t time.Time) error { return errNotImplemented } +func (c *QueuePacketConn) SetReadDeadline(t time.Time) error { return errNotImplemented } +func (c *QueuePacketConn) SetWriteDeadline(t time.Time) error { return errNotImplemented } diff --git a/internal/turbotunnel/remotemap.go b/internal/turbotunnel/remotemap.go new file mode 100644 index 0000000..fecea3f --- /dev/null +++ b/internal/turbotunnel/remotemap.go @@ -0,0 +1,198 @@ +package turbotunnel + +import ( + "container/heap" + "net" + "sync" + "time" +) + +// remoteRecord is a record of a recently seen remote peer, with the time it was +// last seen and queues of outgoing packets. +type remoteRecord struct { + Addr net.Addr + LastSeen time.Time + SendQueue chan []byte + Stash chan []byte +} + +// RemoteMap manages a mapping of live remote peers, keyed by address, to their +// respective send queues. Each peer has two queues: a primary send queue, and a +// "stash". The primary send queue is returned by the SendQueue method. The +// stash is an auxiliary one-element queue accessed using the Stash and Unstash +// methods. The stash is meant for use by callers that need to "unread" a packet +// that's already been removed from the primary send queue. +// +// RemoteMap's functions are safe to call from multiple goroutines. +type RemoteMap struct { + // We use an inner structure to avoid exposing public heap.Interface + // functions to users of remoteMap. + inner remoteMapInner + // Synchronizes access to inner. + lock sync.Mutex + // closed stops the background expiry goroutine. LOCAL FORK ADDITION: + // upstream has no way to stop that goroutine, which leaks it for the + // process lifetime on every RemoteMap created. + closed chan struct{} + closeOnce sync.Once +} + +// NewRemoteMap creates a RemoteMap that expires peers after a timeout. +// +// If the timeout is 0, peers never expire. +// +// The timeout does not have to be kept in sync with smux's idle timeout. If a +// peer is removed from the map while the smux session is still live, the worst +// that can happen is a loss of whatever packets were in the send queue at the +// time. If smux later decides to send more packets to the same peer, we'll +// instantiate a new send queue, and if the peer is ever seen again with a +// matching address, we'll deliver them. +func NewRemoteMap(timeout time.Duration) *RemoteMap { + m := &RemoteMap{ + inner: remoteMapInner{ + byAge: make([]*remoteRecord, 0), + byAddr: make(map[net.Addr]int), + }, + closed: make(chan struct{}), + } + if timeout > 0 { + // LOCAL FORK CHANGE: upstream is `for { time.Sleep(timeout/2); ... }` + // with no exit. Use a ticker and select on m.closed so Close() can stop + // this goroutine. + go func() { + ticker := time.NewTicker(timeout / 2) + defer ticker.Stop() + for { + select { + case <-m.closed: + return + case now := <-ticker.C: + m.lock.Lock() + m.inner.removeExpired(now, timeout) + m.lock.Unlock() + } + } + }() + } + return m +} + +// Close stops the background expiry goroutine started by NewRemoteMap. It is +// safe to call more than once. LOCAL FORK ADDITION. +func (m *RemoteMap) Close() error { + m.closeOnce.Do(func() { close(m.closed) }) + return nil +} + +// SendQueue returns the send queue corresponding to addr, creating it if +// necessary. +func (m *RemoteMap) SendQueue(addr net.Addr) chan []byte { + m.lock.Lock() + defer m.lock.Unlock() + return m.inner.Lookup(addr, time.Now()).SendQueue +} + +// Stash places p in the stash corresponding to addr, if the stash is not +// already occupied. Returns true if the p was placed in the stash, false +// otherwise. +func (m *RemoteMap) Stash(addr net.Addr, p []byte) bool { + m.lock.Lock() + defer m.lock.Unlock() + select { + case m.inner.Lookup(addr, time.Now()).Stash <- p: + return true + default: + return false + } +} + +// Unstash returns the channel that reads from the stash for addr. +func (m *RemoteMap) Unstash(addr net.Addr) <-chan []byte { + m.lock.Lock() + defer m.lock.Unlock() + return m.inner.Lookup(addr, time.Now()).Stash +} + +// remoteMapInner is the inner type of RemoteMap, implementing heap.Interface. +// byAge is the backing store, a heap ordered by LastSeen time, to facilitate +// expiring old records. byAddr is a map from addresses to heap indices, to +// allow looking up by address. Unlike RemoteMap, remoteMapInner requires +// external synchonization. +type remoteMapInner struct { + byAge []*remoteRecord + byAddr map[net.Addr]int +} + +// removeExpired removes all records whose LastSeen timestamp is more than +// timeout in the past. +func (inner *remoteMapInner) removeExpired(now time.Time, timeout time.Duration) { + for len(inner.byAge) > 0 && now.Sub(inner.byAge[0].LastSeen) >= timeout { + record := heap.Pop(inner).(*remoteRecord) + close(record.SendQueue) + } +} + +// Lookup finds the existing record corresponding to addr, or creates a new +// one if none exists yet. It updates the record's LastSeen time and returns the +// record. +func (inner *remoteMapInner) Lookup(addr net.Addr, now time.Time) *remoteRecord { + var record *remoteRecord + i, ok := inner.byAddr[addr] + if ok { + // Found one, update its LastSeen. + record = inner.byAge[i] + record.LastSeen = now + heap.Fix(inner, i) + } else { + // Not found, create a new one. + record = &remoteRecord{ + Addr: addr, + LastSeen: now, + SendQueue: make(chan []byte, QueueSize), + Stash: make(chan []byte, 1), + } + heap.Push(inner, record) + } + return record +} + +// heap.Interface for remoteMapInner. + +func (inner *remoteMapInner) Len() int { + if len(inner.byAge) != len(inner.byAddr) { + panic("inconsistent remoteMap") + } + return len(inner.byAge) +} + +func (inner *remoteMapInner) Less(i, j int) bool { + return inner.byAge[i].LastSeen.Before(inner.byAge[j].LastSeen) +} + +func (inner *remoteMapInner) Swap(i, j int) { + inner.byAge[i], inner.byAge[j] = inner.byAge[j], inner.byAge[i] + inner.byAddr[inner.byAge[i].Addr] = i + inner.byAddr[inner.byAge[j].Addr] = j +} + +func (inner *remoteMapInner) Push(x interface{}) { + record := x.(*remoteRecord) + if _, ok := inner.byAddr[record.Addr]; ok { + panic("duplicate address in remoteMap") + } + // Insert into byAddr map. + inner.byAddr[record.Addr] = len(inner.byAge) + // Insert into byAge slice. + inner.byAge = append(inner.byAge, record) +} + +func (inner *remoteMapInner) Pop() interface{} { + n := len(inner.byAddr) + // Remove from byAge slice. + record := inner.byAge[n-1] + inner.byAge[n-1] = nil + inner.byAge = inner.byAge[:n-1] + // Remove from byAddr map. + delete(inner.byAddr, record.Addr) + return record +} diff --git a/internal/turbotunnel/remotemap_leak_test.go b/internal/turbotunnel/remotemap_leak_test.go new file mode 100644 index 0000000..78a470c --- /dev/null +++ b/internal/turbotunnel/remotemap_leak_test.go @@ -0,0 +1,51 @@ +package turbotunnel + +import ( + "runtime" + "testing" + "time" +) + +// TestQueuePacketConnCloseStopsGoroutine is the regression test for the local +// fork's reason to exist: upstream's RemoteMap expiry goroutine runs forever, +// so creating and discarding many QueuePacketConns (as DNSTT restart does) +// leaks one goroutine each. After Close(), the count must return to baseline. +func TestQueuePacketConnCloseStopsGoroutine(t *testing.T) { + // Let any goroutines from earlier settle. + settle := func() { + for i := 0; i < 50; i++ { + runtime.GC() + time.Sleep(2 * time.Millisecond) + } + } + settle() + base := runtime.NumGoroutine() + + const n = 200 + for i := 0; i < n; i++ { + // Short timeout so the goroutine is definitely started (timeout > 0). + c := NewQueuePacketConn(DummyAddr{}, 50*time.Millisecond) + if err := c.Close(); err != nil { + t.Fatalf("Close returned error: %v", err) + } + } + settle() + + got := runtime.NumGoroutine() + // Allow a small slack for scheduler/runtime goroutines; the key point is we + // are nowhere near base+n (which is what the upstream leak would produce). + if got > base+20 { + t.Fatalf("goroutine leak: baseline=%d after %d create/close cycles=%d (want <= baseline+20)", base, n, got) + } +} + +// TestRemoteMapCloseIdempotent verifies Close can be called repeatedly. +func TestRemoteMapCloseIdempotent(t *testing.T) { + m := NewRemoteMap(10 * time.Millisecond) + if err := m.Close(); err != nil { + t.Fatalf("first Close: %v", err) + } + if err := m.Close(); err != nil { + t.Fatalf("second Close: %v", err) + } +} diff --git a/xray_integration.go b/xray_integration.go index ecd0e69..661ab47 100644 --- a/xray_integration.go +++ b/xray_integration.go @@ -564,10 +564,32 @@ func (m *XrayManager) startNativeStatsFlusher() { defer ticker.Stop() for range ticker.C { m.flushNativeStatsToDB() + m.pruneStaleRuntimeStats() } }) } +// pruneStaleRuntimeStats drops runtime stat entries for clients that have no +// active connections and have been idle well past the online window. Without +// this, statsByEmail keeps one entry per email/UUID that has ever connected and +// never shrinks. The retention comfortably exceeds the online-detection grace +// window so CountOnlineUsers is unaffected; persistent traffic totals live in +// the DB, so dropping the in-memory counter for a long-offline client is safe. +func (m *XrayManager) pruneStaleRuntimeStats() { + retention := 2 * m.onlineWindow() + if retention < 5*time.Minute { + retention = 5 * time.Minute + } + cutoff := time.Now().Add(-retention) + m.statsMu.Lock() + for email, st := range m.statsByEmail { + if st.ActiveConnections <= 0 && (st.LastActive.IsZero() || st.LastActive.Before(cutoff)) { + delete(m.statsByEmail, email) + } + } + m.statsMu.Unlock() +} + func (m *XrayManager) flushNativeStatsToDB() { if statsStore == nil { return diff --git a/xray_native_tuning.go b/xray_native_tuning.go index c83a47f..afc4f24 100644 --- a/xray_native_tuning.go +++ b/xray_native_tuning.go @@ -23,6 +23,14 @@ const ( defaultNativeXHTTPMaxSessions = 16384 defaultNativeXHTTPBufferedPosts = 512 + + // Connected XHTTP sessions are torn down primarily by request-context + // cancellation. This idle timeout is the backstop that reaps a connected + // session whose client vanished without the transport ever reporting it + // (common for XHTTP behind a CDN, where no TCP FIN reaches the origin). + // Matches the SSH idle default so a genuinely idle-but-live tunnel is not + // closed prematurely. + fixedNativeXHTTPIdleMS = 300000 ) var ( @@ -77,3 +85,6 @@ func nativeXHTTPBufferedPostLimit() int { return defaultNativeXHTTPBufferedPosts func nativeMuxUDPIdleTimeout() time.Duration { return fixedNativeMuxUDPIdleMS * time.Millisecond } +func nativeXHTTPIdleTimeout() time.Duration { + return fixedNativeXHTTPIdleMS * time.Millisecond +} diff --git a/xray_xhttp.go b/xray_xhttp.go index fdaa5af..6b714bb 100644 --- a/xray_xhttp.go +++ b/xray_xhttp.go @@ -117,11 +117,66 @@ func (ib *nativeInbound) serveXHTTPListener(ln net.Listener) { srv.TLSConfig = ib.tlsConfig _ = http2.ConfigureServer(srv, h2s) } + // Backstop reaper for connected sessions whose client vanished without the + // request context ever firing. Lives for the lifetime of this listener. + stopSweep := make(chan struct{}) + defer close(stopSweep) + xrayGo(fmt.Sprintf("native xray XHTTP idle sweeper inbound=%q", ib.tag), func() { ib.sweepXHTTPSessions(stopSweep) }) if err := srv.Serve(ln); err != nil && !errors.Is(err, http.ErrServerClosed) && !isListenerClosed(err) { xrayLogf("native xray: XHTTP server for inbound %q stopped: %v", ib.tag, err) } } +// sweepXHTTPSessions periodically evicts connected XHTTP sessions that have seen +// no traffic (in either direction) within the idle timeout. This is the safety +// net behind the per-request context watch in handleXHTTPDownload; it only ever +// touches sessions whose lastSeen has genuinely gone stale, so an active tunnel +// (which refreshes lastSeen via nativeXHTTPConn.onActivity) is never reaped. +func (ib *nativeInbound) sweepXHTTPSessions(stop <-chan struct{}) { + defer xrayRecover(fmt.Sprintf("native xray XHTTP idle sweeper inbound=%q", ib.tag)) + idle := nativeXHTTPIdleTimeout() + if idle <= 0 { + return + } + interval := idle / 4 + if interval < 15*time.Second { + interval = 15 * time.Second + } + t := time.NewTicker(interval) + defer t.Stop() + for { + select { + case <-stop: + return + case <-t.C: + ib.reapStaleXHTTPSessions(idle) + } + } +} + +func (ib *nativeInbound) reapStaleXHTTPSessions(idle time.Duration) { + now := time.Now() + var stale []*nativeXHTTPSession + ib.xhttpMu.Lock() + for id, s := range ib.xhttpSessions { + s.mu.Lock() + connected := s.connected + last := s.lastSeen + s.mu.Unlock() + // Unconnected sessions have their own 30s reaper; only reap connected + // ones that have gone idle past the timeout. + if connected && now.Sub(last) >= idle { + delete(ib.xhttpSessions, id) + stale = append(stale, s) + } + } + ib.xhttpMu.Unlock() + for _, s := range stale { + xrayTracef("native xray: xhttp idle sweep closing session=%q inbound=%q", s.id, ib.tag) + s.close() + } +} + func (ib *nativeInbound) xhttpServerMaxHeaderBytes() int { if ib.xhttpMaxHeaderBytes > 0 { return ib.xhttpMaxHeaderBytes @@ -636,16 +691,32 @@ func (ib *nativeInbound) handleXHTTPDownload(w http.ResponseWriter, r *http.Requ var reader io.Reader = sess.queue resp := newNativeXHTTPResponseWriter(w) xc := &nativeXHTTPConn{ - reader: reader, - writer: resp, - remote: remote, - local: dummyLocalAddr(r), + reader: reader, + writer: resp, + remote: remote, + local: dummyLocalAddr(r), + onActivity: sess.touch, } xc.onClose = func() { resp.close() sess.close() } + // When the download GET is cancelled (client gone, or a CDN closes the + // origin stream after its own idle timeout) the HTTP request context fires. + // Closing xc unblocks the tunnel's uplink reader (via sess.close -> queue + // close) and closes the backend, so handleXHTTPDownload returns and its + // deferred deleteXHTTPSession runs. Without this watch an idle tunnel whose + // client vanished silently would never be torn down. The goroutine exits on + // sess.done once the session closes for any reason. + go func() { + select { + case <-r.Context().Done(): + _ = xc.Close() + case <-sess.done: + } + }() + ib.dispatchXHTTPConn(xc, remote) _ = xc.Close() } @@ -727,6 +798,11 @@ type nativeXHTTPConn struct { closeOnce sync.Once onClose func() + // onActivity, when set, is called after any successful read or write so the + // owning session's lastSeen reflects real bidirectional traffic (not just + // HTTP request arrivals). The idle sweeper relies on this to avoid reaping a + // tunnel that is actively streaming in only one direction. + onActivity func() } func (c *nativeXHTTPConn) Read(p []byte) (int, error) { @@ -736,10 +812,20 @@ func (c *nativeXHTTPConn) Read(p []byte) (int, error) { c.deadlineMu.Unlock() _ = dr.SetReadDeadline(d) } - return c.reader.Read(p) + n, err := c.reader.Read(p) + if n > 0 && c.onActivity != nil { + c.onActivity() + } + return n, err } -func (c *nativeXHTTPConn) Write(p []byte) (int, error) { return c.writer.Write(p) } +func (c *nativeXHTTPConn) Write(p []byte) (int, error) { + n, err := c.writer.Write(p) + if n > 0 && c.onActivity != nil { + c.onActivity() + } + return n, err +} func (c *nativeXHTTPConn) Close() error { c.closeOnce.Do(func() {