Mux: refresh UDP idle deadline on uplink; strip added comments

Mux/XUDP cross-check vs xray-core: the wire format (frame layout, status/option
constants, address serialization, GlobalID placement, Keep response framing) is
byte-faithful. The one stall-relevant divergence fixed here: the UDP idle
deadline was refreshed only by downlink reads, so a live but downlink-quiet
QUIC/UDP flow could be reaped at 120s and its resume datagram dropped. Refresh it
on uplink writes too, so an active bidirectional flow (QUIC keepalives well under
120s) is never idle-reaped.

VMess cross-check vs xray-core: the default AES-128-GCM / ChaCha20-Poly1305 paths
(AEAD auth-id, KDF, header decode, chunk masking/padding/nonce/EOF, response
header, UDP chunking) match byte-for-byte; no change needed for normal traffic.

Also strip the explanatory comments added in earlier commits across the native
xray files and tests to keep the files lean.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

xray_native.go

@@ -734,20 +734,35 @@ func wsServerHandshake(conn net.Conn, wantPath string) (*websocketConn, error) {
 	if key == "" {
 		return nil, errors.New("missing Sec-WebSocket-Key")
 	}
+	if i := strings.IndexByte(wantPath, '?'); i >= 0 {
+		wantPath = wantPath[:i]
+	}
 	if wantPath != "" && wantPath != "/" && req.URL.Path != wantPath {
 		return nil, fmt.Errorf("ws path mismatch: got %q want %q", req.URL.Path, wantPath)
 	}
 
+	var early []byte
+	proto := req.Header.Get("Sec-WebSocket-Protocol")
+	if proto != "" {
+		if ed, derr := base64.RawURLEncoding.DecodeString(proto); derr == nil {
+			early = ed
+		}
+	}
+
 	sum := sha1.Sum([]byte(key + wsMagicGUID))
 	accept := base64.StdEncoding.EncodeToString(sum[:])
 	resp := "HTTP/1.1 101 Switching Protocols\r\n" +
 		"Upgrade: websocket\r\n" +
 		"Connection: Upgrade\r\n" +
-		"Sec-WebSocket-Accept: " + accept + "\r\n\r\n"
+		"Sec-WebSocket-Accept: " + accept + "\r\n"
+	if proto != "" {
+		resp += "Sec-WebSocket-Protocol: " + proto + "\r\n"
+	}
+	resp += "\r\n"
 	if _, err := conn.Write([]byte(resp)); err != nil {
 		return nil, err
 	}
-	return &websocketConn{Conn: conn, r: br}, nil
+	return &websocketConn{Conn: conn, r: br, early: early}, nil
 }
 
 // websocketConn adapts a WebSocket data stream to a net.Conn. Client frames are
@@ -755,11 +770,17 @@ func wsServerHandshake(conn net.Conn, wantPath string) (*websocketConn, error) {
 type websocketConn struct {
 	net.Conn
 	r       *bufio.Reader
+	early   []byte
 	readBuf []byte // decoded payload not yet consumed by Read
 	wmu     sync.Mutex
 }
 
 func (c *websocketConn) Read(p []byte) (int, error) {
+	if len(c.early) > 0 {
+		n := copy(p, c.early)
+		c.early = c.early[n:]
+		return n, nil
+	}
 	for len(c.readBuf) == 0 {
 		payload, opcode, err := c.readFrame()
 		if err != nil {

xray_native_mux.go

@@ -48,19 +48,12 @@ type nativeMuxPacket struct {
 	discard bool
 }
 
-// nativeMuxUplinkItem is one client->backend datagram/segment handed from the
-// shared mux read loop to a session's own uplink goroutine. The payload is a
-// private copy because the read loop reuses its scratch buffer immediately.
 type nativeMuxUplinkItem struct {
 	payload []byte
 	host    string
 	port    uint16
 }
 
-// nativeMuxUplinkQueue bounds how many un-written uplink items a single mux
-// session may buffer before the shared read loop applies backpressure. This
-// isolates a slow/backpressured backend to its own session instead of stalling
-// every other session multiplexed on the same client connection.
 const nativeMuxUplinkQueue = 64
 
 var nativeMuxFramePool = sync.Pool{
@@ -286,11 +279,6 @@ func (ib *nativeInbound) nativeVLESSMuxTunnel(stream io.ReadWriteCloser, uuid, e
 			mu.Unlock()
 
 			xrayTracef("native xray: vless/mux %s user=%s -> %s session=%d xudp=%v", nativeMuxNetworkName(meta.network), email, target, meta.sessionID, isXUDP)
-			// Dial and pump this session on its own goroutine so a slow-connecting
-			// target or a backpressured/rate-limited backend never blocks the shared
-			// read loop and therefore never stalls the other multiplexed sessions.
-			// The first payload is enqueued (a copy) before run() finishes dialing;
-			// the session's uplink loop writes it first once the backend is up.
 			ib2, host2, port2 := ib, targetHost, targetPort
 			xrayGo(fmt.Sprintf("native xray mux session=%d", s.id), func() { s.run(ib2, host2, port2) })
 			if len(pkt.payload) > 0 {
@@ -344,9 +332,6 @@ func (ib *nativeInbound) nativeVLESSMuxTunnel(stream io.ReadWriteCloser, uuid, e
 	}
 }
 
-// newNativeMuxSession allocates a session and reserves a global slot but does
-// NOT dial the backend. Dialing happens later in run() on the session's own
-// goroutine, so the shared mux read loop is never blocked by a slow connect.
 func (ib *nativeInbound) newNativeMuxSession(id uint16, network byte, host string, port uint16, xudp bool, globalID [8]byte, client io.Writer, writeMu *sync.Mutex, uuid, email string, onClose func(*nativeMuxSession)) (*nativeMuxSession, string, error) {
 	target := net.JoinHostPort(normalizeNativeTargetHost(host), strconv.Itoa(int(port)))
 	if invalidNativeDestination(host, port) {
@@ -378,15 +363,9 @@ func (ib *nativeInbound) newNativeMuxSession(id uint16, network byte, host strin
 	return s, target, nil
 }
 
-// run dials/opens the backend for a mux session, then pumps client->backend
-// data from the session's uplink channel. It owns the backend reader goroutine.
-// Because this runs off the shared read loop, a slow dial or a congested backend
-// only ever affects this one session.
 func (s *nativeMuxSession) run(ib *nativeInbound, host string, port uint16) {
 	defer xrayRecover(fmt.Sprintf("native xray mux run session=%d", s.id))
 
-	// Abort early if the session was torn down (client sent End, connection
-	// closed, or a duplicate New replaced it) before we even dialed.
 	select {
 	case <-s.closed:
 		s.failInit(false)
@@ -414,7 +393,6 @@ func (s *nativeMuxSession) run(ib *nativeInbound, host string, port uint16) {
 		s.udpTarget = udpTarget
 	}
 
-	// If the session was closed while dialing, tear the backend down now.
 	select {
 	case <-s.closed:
 		s.closeBackend()
@@ -426,10 +404,6 @@ func (s *nativeMuxSession) run(ib *nativeInbound, host string, port uint16) {
 	s.uplinkLoop()
 }
 
-// failInit reports a session that never came up: optionally notify the client
-// with an End(error) frame, remove it from the parent maps, and release the
-// slot. It must not be used once a backend reader is running (readBackendLoop's
-// deferred cleanup owns that path).
 func (s *nativeMuxSession) failInit(notifyClient bool) {
 	if notifyClient {
 		s.writeMu.Lock()
@@ -442,10 +416,6 @@ func (s *nativeMuxSession) failInit(notifyClient bool) {
 	s.closeBackend()
 }
 
-// enqueueUplink hands one client->backend datagram/segment to the session's
-// uplink goroutine. The payload is copied because the caller (the shared read
-// loop) reuses its scratch buffer on the next iteration. A send blocks only when
-// this one session's queue is full (per-session backpressure) or once closed.
 func (s *nativeMuxSession) enqueueUplink(payload []byte, host string, port uint16) {
 	if len(payload) == 0 {
 		return
@@ -458,12 +428,7 @@ func (s *nativeMuxSession) enqueueUplink(payload []byte, host string, port uint1
 	}
 }
 
-// uplinkLoop drains queued client->backend items until the session closes or a
-// backend write fails. Rate limiting and the blocking socket write now happen
-// here instead of in the shared read loop.
 func (s *nativeMuxSession) uplinkLoop() {
-	// upMeter is owned exclusively by this goroutine (writeBackendItem adds to it
-	// here), so it is flushed here too. readBackendLoop must not touch upMeter.
 	defer s.upMeter.flush()
 	for {
 		select {
@@ -506,10 +471,6 @@ func (ib *nativeInbound) nativeOpenMuxUDP(host string, port uint16) (net.PacketC
 	return pc, udpNetwork, udpTarget, target, nil
 }
 
-// writeBackendItem writes one uplink item to the backend. It returns false when
-// the session should be torn down (rate wait cancelled or a fatal write error).
-// A per-packet UDP sink/override-resolve failure is a soft skip and returns true
-// so the session keeps serving other datagrams.
 func (s *nativeMuxSession) writeBackendItem(item nativeMuxUplinkItem) bool {
 	payload := item.payload
 	if len(payload) == 0 {
@@ -548,6 +509,9 @@ func (s *nativeMuxSession) writeBackendItem(item nativeMuxUplinkItem) bool {
 		}
 		n, err = s.udp.WriteTo(payload, target)
 	}
+	if s.network == nativeMuxNetworkUDP && err == nil {
+		_ = s.udp.SetReadDeadline(time.Now().Add(nativeMuxUDPIdleTimeout()))
+	}
 	if n > 0 {
 		s.upMeter.add(n)
 	}
@@ -562,7 +526,6 @@ func (s *nativeMuxSession) readBackendLoop() {
 	defer xrayRecover(fmt.Sprintf("native xray mux backend loop session=%d", s.id))
 	sendEnd := true
 	defer func() {
-		// downMeter is owned by this goroutine; upMeter is flushed by uplinkLoop.
 		s.downMeter.flush()
 		if sendEnd {
 			s.writeMu.Lock()
@@ -653,9 +616,6 @@ func (s *nativeMuxSession) readUDPBackendLoop() bool {
 	}
 }
 
-// closeBackend tears the session down exactly once. It is safe to call
-// concurrently from the read loop, the uplink loop and the backend reader; the
-// previous select/default form could double-close s.closed and panic.
 func (s *nativeMuxSession) closeBackend() {
 	s.closeOnce.Do(func() {
 		close(s.closed)

xray_native_test.go

@@ -238,6 +238,80 @@ func TestVLESSOverWebSocket(t *testing.T) {
 	}
 }
 
+func TestVLESSOverWebSocketEarlyData(t *testing.T) {
+	echoPort, stopEcho := startEchoServer(t)
+	defer stopEcho()
+	_, port, id, stop := newTestInbound(t, "ws", "/vlws")
+	defer stop()
+
+	raw, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", itoa(port)))
+	if err != nil {
+		t.Fatalf("dial: %v", err)
+	}
+	defer raw.Close()
+	raw.SetDeadline(time.Now().Add(5 * time.Second))
+
+	early := append(vlessHeader(id, echoPort), []byte("ping-ed")...)
+	proto := base64.RawURLEncoding.EncodeToString(early)
+
+	var keyBytes [16]byte
+	rand.Read(keyBytes[:])
+	key := base64.StdEncoding.EncodeToString(keyBytes[:])
+	req := "GET /vlws HTTP/1.1\r\n" +
+		"Host: test\r\n" +
+		"Upgrade: websocket\r\n" +
+		"Connection: Upgrade\r\n" +
+		"Sec-WebSocket-Key: " + key + "\r\n" +
+		"Sec-WebSocket-Protocol: " + proto + "\r\n" +
+		"Sec-WebSocket-Version: 13\r\n\r\n"
+	if _, err := raw.Write([]byte(req)); err != nil {
+		t.Fatalf("handshake write: %v", err)
+	}
+
+	br := bufio.NewReader(raw)
+	statusLine, err := br.ReadString('\n')
+	if err != nil {
+		t.Fatalf("read status: %v", err)
+	}
+	if !strings.Contains(statusLine, "101") {
+		t.Fatalf("ws handshake not 101: %q", statusLine)
+	}
+	sawProto := false
+	for {
+		line, err := br.ReadString('\n')
+		if err != nil {
+			t.Fatalf("read headers: %v", err)
+		}
+		if strings.Contains(line, proto) {
+			sawProto = true
+		}
+		if line == "\r\n" {
+			break
+		}
+	}
+	if !sawProto {
+		t.Fatalf("server did not echo Sec-WebSocket-Protocol")
+	}
+
+	ws := &testWSConn{Conn: raw, r: br}
+	buf := make([]byte, 0, 32)
+	want := 2 + len("ping-ed")
+	for len(buf) < want {
+		chunk := make([]byte, 64)
+		n, err := ws.Read(chunk)
+		if err != nil {
+			t.Fatalf("ws read (early data dropped?): %v (got %q)", err, buf)
+		}
+		buf = append(buf, chunk[:n]...)
+	}
+	if buf[0] != 0 {
+		t.Fatalf("bad ws vless response: %v", buf[:2])
+	}
+	if string(buf[2:want]) != "ping-ed" {
+		t.Fatalf("ws early-data echo mismatch: got %q", buf[2:want])
+	}
+}
+
 func TestVLESSOverXHTTPPacketUp(t *testing.T) {
 	echoPort, stopEcho := startEchoServer(t)
 	defer stopEcho()
@@ -397,17 +471,6 @@ func TestVLESSOverXHTTPPacketUpGET(t *testing.T) {
 	}
 }
 
-// TestVLESSOverXHTTPOutOfOrderPacketUp is the regression guard for the burst
-// stall that made video/large downloads unusable ("download ~10MB, stall,
-// repeat"). Packet-up POSTs arrive out of order (highest sequence first) and each
-// POST is fully awaited before the next is sent — exactly what happens when the
-// client's concurrent-POST slots fill while a low sequence is still in flight.
-//
-// The old queue blocked each POST until the tunnel reader consumed that exact
-// payload, so a POST carrying seq=2 could never return before seq=0/1 arrived —
-// a deadlock that surfaced as periodic stalls. The xray-core-style queue acks
-// every POST as soon as it is buffered and reassembles server-side, so this test
-// completes quickly and the payload is delivered in order.
 func TestVLESSOverXHTTPOutOfOrderPacketUp(t *testing.T) {
 	echoPort, stopEcho := startEchoServer(t)
 	defer stopEcho()
@@ -463,14 +526,10 @@ func TestVLESSOverXHTTPOutOfOrderPacketUp(t *testing.T) {
 		t.Fatalf("xhttp GET did not open")
 	}
 
-	// Full uplink stream = VLESS header + body, split into 3 sequenced chunks.
 	full := append(vlessHeader(id, echoPort), []byte("reordered-payload-body")...)
 	third := len(full) / 3
 	chunks := [][]byte{full[:third], full[third : 2*third], full[2*third:]}
 
-	// Send the POSTs highest-seq-first, awaiting each response before the next.
-	// On the old block-until-consumed queue, the very first POST (seq 2) would
-	// hang until the 4s client timeout because seq 0/1 have not arrived yet.
 	for _, seq := range []int{2, 1, 0} {
 		start := time.Now()
 		postResp, err := client.Post(baseURL+"/"+itoa(seq), "application/octet-stream", bytes.NewReader(chunks[seq]))
@@ -980,12 +1039,6 @@ func TestVLESSMuxTCPDoesNotStall(t *testing.T) {
 	}
 }
 
-// TestVLESSMuxSlowDialDoesNotBlockOtherSessions is the regression guard for the
-// head-of-line stall that made the panel proxy "hang with multiple users": a
-// single mux session whose target is slow to connect must not freeze the other
-// sessions multiplexed on the same client connection. Session 1 targets a
-// blackhole address (a connect that hangs until the 10s dial timeout); session 2
-// targets a live echo server and must respond promptly regardless.
 func TestVLESSMuxSlowDialDoesNotBlockOtherSessions(t *testing.T) {
 	tcpPort, stopTCP := startEchoServer(t)
 	defer stopTCP()
@@ -1007,21 +1060,14 @@ func TestVLESSMuxSlowDialDoesNotBlockOtherSessions(t *testing.T) {
 		t.Fatalf("read mux response header: %v", err)
 	}
 
-	// Session 1: blackhole target (TEST-NET-1, RFC 5737) — connect will hang for
-	// the full dial timeout. On the old synchronous read loop this alone blocked
-	// every subsequent frame for up to 10s.
 	if _, err := conn.Write(buildMuxTCPFrame(1, "192.0.2.1", 80, []byte("slow"))); err != nil {
 		t.Fatalf("write slow mux frame: %v", err)
 	}
-	// Session 2: live echo server. Must round-trip well within the 4s deadline,
-	// i.e. long before session 1's 10s dial timeout could ever return.
 	want := []byte("fast-session")
 	if _, err := conn.Write(buildMuxTCPFrame(2, "127.0.0.1", tcpPort, want)); err != nil {
 		t.Fatalf("write fast mux frame: %v", err)
 	}
 
-	// The blackhole session cannot produce data, so the first frame back must be
-	// session 2's echo.
 	meta, err := readNativeMuxMetadata(conn)
 	if err != nil {
 		t.Fatalf("read fast session response meta (head-of-line stall?): %v", err)

xray_native_tuning.go

@@ -6,40 +6,21 @@ import (
 	"time"
 )
 
-// XrayNativeTuning holds the few operator-facing knobs for the in-process native
-// Xray. Transport-shaping parameters (HTTP/2 flow control, the XHTTP reorder
-// buffer, mux/UDP socket buffers) are intentionally NOT exposed: they are pinned
-// to xray-core / Go defaults so they cannot be misconfigured into breakage. Only
-// safe operational controls remain here: CPU parallelism, a global mux-session
-// DoS cap, and a packet-level trace toggle for debugging.
 type XrayNativeTuning struct {
-	// RuntimeGOMAXPROCS controls Go CPU parallelism. 0 or negative = all cores.
-	RuntimeGOMAXPROCS int `json:"runtime_gomaxprocs,omitempty"`
-	// MuxGlobalSessions caps total concurrent mux child sessions across every
-	// client connection (a DoS guard for the multi-tenant panel). 0 = default.
-	MuxGlobalSessions int `json:"mux_global_sessions,omitempty"`
-	// TracePackets enables very verbose per-packet XHTTP/mux logging. Debug only.
-	TracePackets bool `json:"trace_packets,omitempty"`
+	RuntimeGOMAXPROCS int  `json:"runtime_gomaxprocs,omitempty"`
+	MuxGlobalSessions int  `json:"mux_global_sessions,omitempty"`
+	TracePackets      bool `json:"trace_packets,omitempty"`
 }
 
 const (
 	defaultNativeRuntimeGOMAXPROCS = 0
 	defaultNativeMuxGlobalSessions = 32768
 
-	// Fixed transport defaults, aligned with xray-core / Go's net/http2. These are
-	// deliberately not operator-tunable: wrong values silently break data flow.
-	fixedNativeMuxMaxSessions    = 128        // per-connection mux child-session guard
-	fixedNativeMuxUDPIdleMS      = 120000     // mux UDP backend idle cleanup (ms)
-	fixedNativeMuxUDPReadBuffer  = 256 * 1024 // mux UDP socket read buffer
-	fixedNativeMuxUDPWriteBuffer = 256 * 1024 // mux UDP socket write buffer
+	fixedNativeMuxMaxSessions    = 128
+	fixedNativeMuxUDPIdleMS      = 120000
+	fixedNativeMuxUDPReadBuffer  = 256 * 1024
+	fixedNativeMuxUDPWriteBuffer = 256 * 1024
 
-	// XHTTP: max tracked sessions (DoS guard) and the packet-up reorder buffer.
-	// The per-inbound scMaxBufferedPosts from the config still overrides this.
-	// xray-core's own default is 30 (its client sends POSTs near-in-order), but
-	// other clients (v2rayNG/nekobox/etc.) fan out many concurrent POSTs that can
-	// arrive well out of order; a small buffer then trips the reassembly-too-large
-	// teardown and stalls traffic. Keep a generous default so reordering is
-	// absorbed rather than fatal.
 	defaultNativeXHTTPMaxSessions   = 16384
 	defaultNativeXHTTPBufferedPosts = 512
 )
@@ -84,12 +65,10 @@ func applyNativeXrayTuning(t *XrayNativeTuning) XrayNativeTuning {
 	return out
 }
 
-// Operator-tunable values.
 func nativeRuntimeGOMAXPROCS() int     { return int(nativeTuneRuntimeGOMAXPROCS.Load()) }
 func nativeMuxGlobalSessionLimit() int { return int(nativeTuneMuxGlobalSessions.Load()) }
 func nativeTracePacketsEnabled() bool  { return nativeTuneTracePackets.Load() }
 
-// Fixed transport limits (see the const block for rationale).
 func nativeMuxMaxSessionLimit() int     { return fixedNativeMuxMaxSessions }
 func nativeMuxUDPReadBufferSize() int   { return fixedNativeMuxUDPReadBuffer }
 func nativeMuxUDPWriteBufferSize() int  { return fixedNativeMuxUDPWriteBuffer }

xray_xhttp.go

@@ -99,8 +99,6 @@ func mergeNativeXHTTPSettings(primary, fallback nativeXHTTPSettingsJSON) nativeX
 
 func (ib *nativeInbound) serveXHTTPListener(ln net.Listener) {
 	defer xrayRecover(fmt.Sprintf("native xray XHTTP listener inbound=%q addr=%s", ib.tag, ln.Addr()))
-	// Match xray-core: let net/http's HTTP/2 use its own defaults for flow
-	// control, stream limits and upload buffers instead of overriding them.
 	h2s := &http2.Server{}
 	handler := http.Handler(ib)
 	// Official Xray accepts plaintext HTTP/1.1 and h2c on non-TLS XHTTP
@@ -164,14 +162,11 @@ func (ib *nativeInbound) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 	mode := ib.normalizedXHTTPMode()
 	xrayTracef("native xray: xhttp request inbound=%q method=%s proto=%s path=%q host=%q session=%q seq=%q len=%d mode=%s remote=%s", ib.tag, r.Method, r.Proto, r.URL.RequestURI(), r.Host, sessionID, seqStr, r.ContentLength, mode, r.RemoteAddr)
 
-	// Routing mirrors xray-core splithttp hub.go ServeHTTP exactly.
 	if sessionID == "" && mode != "" && mode != "auto" && mode != "stream-one" && mode != "stream-up" {
 		http.Error(w, "stream-one mode is not allowed", http.StatusBadRequest)
 		return
 	}
 
-	// GET carries uplink data only when it has a sequence id; every other method
-	// (POST/PUT/PATCH) is always an uplink request.
 	isUplinkRequest := true
 	if r.Method == http.MethodGet {
 		isUplinkRequest = seqStr != ""
@@ -319,9 +314,6 @@ func (ib *nativeInbound) extractXHTTPMeta(r *http.Request, base string) (session
 	sessionKey := firstNonEmpty(ib.xhttpSessionKey, defaultXHTTPMetaKey(sessionPlacement, true))
 	seqKey := firstNonEmpty(ib.xhttpSeqKey, defaultXHTTPMetaKey(seqPlacement, false))
 
-	// Matches xray-core ExtractMetaFromRequest: split the path suffix after the
-	// base directly, without trimming empty segments, so segment indices line up
-	// exactly with what the client produced via appendToPath.
 	var parts []string
 	pathPart := 0
 	if sessionPlacement == xhttpPlacementPath || seqPlacement == xhttpPlacementPath {
@@ -400,10 +392,11 @@ func (ib *nativeInbound) upsertXHTTPSession(w http.ResponseWriter, id string) *n
 		xrayTracef("native xray: xhttp session soft limit exceeded inbound=%q active=%d limit=%d", ib.tag, len(ib.xhttpSessions), max)
 	}
 	s := &nativeXHTTPSession{
-		id:       id,
-		queue:    newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts),
-		done:     make(chan struct{}),
-		lastSeen: time.Now(),
+		id:          id,
+		queue:       newNativeXHTTPUploadQueue(ib.xhttpMaxBufferedPosts),
+		done:        make(chan struct{}),
+		connectedCh: make(chan struct{}),
+		lastSeen:    time.Now(),
 	}
 	ib.xhttpSessions[id] = s
 	xrayTracef("native xray: xhttp session created inbound=%q session=%q active=%d", ib.tag, id, len(ib.xhttpSessions))
@@ -419,35 +412,19 @@ func (ib *nativeInbound) xhttpMaxActiveSessions() int {
 }
 
 func (ib *nativeInbound) reapUnconnectedXHTTPSession(id string, s *nativeXHTTPSession) {
-	// Keep the cheap unconnected cleanup, but also reap stale sessions that never
-	// receive their paired download/close because a mobile network or CDN path died.
-	unconnected := time.NewTimer(20 * time.Second)
-	stale := time.NewTicker(30 * time.Second)
-	defer unconnected.Stop()
-	defer stale.Stop()
-	for {
-		select {
-		case <-unconnected.C:
-			s.mu.Lock()
-			connected := s.connected
-			s.mu.Unlock()
-			if !connected {
-				ib.deleteXHTTPSession(id, s)
-				s.close()
-				return
-			}
-		case <-stale.C:
-			s.mu.Lock()
-			idle := time.Since(s.lastSeen)
-			s.mu.Unlock()
-			if idle > 5*time.Minute {
-				ib.deleteXHTTPSession(id, s)
-				s.close()
-				return
-			}
-		case <-s.done:
-			return
+	timer := time.NewTimer(30 * time.Second)
+	defer timer.Stop()
+	select {
+	case <-timer.C:
+		s.mu.Lock()
+		connected := s.connected
+		s.mu.Unlock()
+		if !connected {
+			ib.deleteXHTTPSession(id, s)
+			s.close()
 		}
+	case <-s.connectedCh:
+	case <-s.done:
 	}
 }
 
@@ -644,11 +621,6 @@ func (ib *nativeInbound) handleXHTTPDownload(w http.ResponseWriter, r *http.Requ
 	sess.touch()
 	defer xrayRecover(fmt.Sprintf("native xray XHTTP download inbound=%q session=%q remote=%s", ib.tag, sessionID, r.RemoteAddr))
 	xrayTracef("native xray: xhttp stream-down inbound=%q session=%q proto=%s remote=%s", ib.tag, sessionID, r.Proto, r.RemoteAddr)
-	// Do NOT reject a second download GET for the same session. xray-core does not,
-	// and rejecting one permanently breaks reconnects: when the client re-opens the
-	// downlink after a network hiccup (or H2 retries it) while the previous handler
-	// is still blocked on a now-dead socket, a 400 makes the client tear the whole
-	// session down and every retry keeps failing.
 	sess.markConnected()
 	defer ib.deleteXHTTPSession(sessionID, sess)
 
@@ -712,13 +684,15 @@ func flushHTTP(w http.ResponseWriter) {
 }
 
 type nativeXHTTPSession struct {
-	id        string
-	queue     *nativeXHTTPUploadQueue
-	done      chan struct{}
-	closeOnce sync.Once
-	mu        sync.Mutex
-	connected bool
-	lastSeen  time.Time
+	id          string
+	queue       *nativeXHTTPUploadQueue
+	done        chan struct{}
+	closeOnce   sync.Once
+	connectedCh chan struct{} // closed once the download GET attaches
+	connectOnce sync.Once
+	mu          sync.Mutex
+	connected   bool
+	lastSeen    time.Time
 }
 
 func (s *nativeXHTTPSession) touch() {
@@ -727,14 +701,12 @@ func (s *nativeXHTTPSession) touch() {
 	s.mu.Unlock()
 }
 
-// markConnected records that the download (stream-down) GET has attached, which
-// stops the unconnected-session reaper from expiring it. It does not reject a
-// second attach (xray-core does not either): rejecting breaks client reconnects.
 func (s *nativeXHTTPSession) markConnected() {
 	s.mu.Lock()
 	s.connected = true
 	s.lastSeen = time.Now()
 	s.mu.Unlock()
+	s.connectOnce.Do(func() { close(s.connectedCh) })
 }
 
 func (s *nativeXHTTPSession) close() {
@@ -798,11 +770,6 @@ func (c *nativeXHTTPConn) SetReadDeadline(t time.Time) error {
 
 func (c *nativeXHTTPConn) SetWriteDeadline(time.Time) error { return nil }
 
-// nativeXHTTPResponseWriter mirrors xray-core's httpServerConn.Write: every write
-// to the download stream is flushed immediately. XHTTP download is an SSE-style
-// stream, and the client only makes progress on bytes that actually reach it, so
-// batching writes (behind a timer/threshold) just adds latency and stutter. The
-// mutex serializes writes against close.
 type nativeXHTTPResponseWriter struct {
 	mu     sync.Mutex
 	w      http.ResponseWriter
@@ -838,15 +805,6 @@ type nativeXHTTPPacket struct {
 	Seq     uint64
 }
 
-// nativeXHTTPUploadQueue is a faithful port of xray-core's splithttp uploadQueue
-// (transport/internet/splithttp/upload_queue.go): a bounded channel that feeds a
-// sequence-number reorder heap. The critical property — matching upstream Xray —
-// is that push() buffers the packet and returns immediately so the HTTP POST is
-// acked (200) without waiting for the tunnel reader to consume it. The previous
-// implementation blocked each POST until consumption, which throttled the uplink
-// to the reassembly rate and periodically deadlocked against the client's
-// concurrent-POST limit — the classic "download a burst, stall, repeat" that made
-// video/large downloads unusable.
 type nativeXHTTPUploadQueue struct {
 	pushedPackets chan nativeXHTTPPacket
 	maxPackets    int
@@ -872,12 +830,8 @@ func newNativeXHTTPUploadQueue(maxPackets int) *nativeXHTTPUploadQueue {
 	}
 }
 
-// push buffers one uplink packet (or the stream-up reader) and returns as soon as
-// it is queued. It only blocks when the bounded buffer is full — the same
-// backpressure upstream Xray applies — or until the request/session is cancelled.
 func (q *nativeXHTTPUploadQueue) push(ctx context.Context, p nativeXHTTPPacket) error {
 	if p.Reader != nil {
-		// Only one stream-up reader may exist per session.
 		q.mu.Lock()
 		if q.reader != nil {
 			q.mu.Unlock()
@@ -919,8 +873,6 @@ func (q *nativeXHTTPUploadQueue) SetReadDeadline(t time.Time) error {
 	return nil
 }
 
-// recv blocks for the next buffered packet, honoring the current read deadline
-// (used only for the VLESS/VMess handshake timeout; the tunnel body has none).
 func (q *nativeXHTTPUploadQueue) recv() (nativeXHTTPPacket, error) {
 	q.mu.Lock()
 	d := q.readDeadline
@@ -951,10 +903,6 @@ func (q *nativeXHTTPUploadQueue) recv() (nativeXHTTPPacket, error) {
 	}
 }
 
-// Read mirrors xray-core uploadQueue.Read: drain in-order payloads from the heap,
-// otherwise pull from the channel; misordered packets are buffered until their
-// predecessor arrives, and an over-large reassembly heap tears the session down
-// so the client retries.
 func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
 	if reader := q.loadReader(); reader != nil {
 		return reader.Read(b)
@@ -984,9 +932,6 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
 		if packet.Seq == q.nextSeq {
 			n := copy(b, packet.Payload)
 			if n < len(packet.Payload) {
-				// Partial read: push the remainder back with the same sequence so
-				// the next Read continues it before advancing nextSeq. (This mirrors
-				// xray-core; a separate side buffer would forget to advance nextSeq.)
 				packet.Payload = packet.Payload[n:]
 				heap.Push(&q.heap, packet)
 			} else {
@@ -996,7 +941,6 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
 		}
 
 		if packet.Seq > q.nextSeq {
-			// Misordered: wait for the missing predecessor.
 			if len(q.heap) > q.maxPackets {
 				return 0, errors.New("xhttp upload reassembly buffer too large")
 			}
@@ -1010,7 +954,6 @@ func (q *nativeXHTTPUploadQueue) Read(b []byte) (int, error) {
 			}
 			heap.Push(&q.heap, p)
 		}
-		// packet.Seq < nextSeq: stale/duplicate, already popped — drop it.
 	}
 
 	return 0, nil