Hysteria client: Reset stalled writes when the connection is closed (#7172)

https://github.com/XTLS/Xray-core/pull/7172#issuecomment-6208589708

Fixes https://github.com/XTLS/Xray-core/issues/7169#issuecomment-6204450876

---------

Co-authored-by: LjhAUMEM <llnu14702@gmail.com>

transport/internet/hysteria/conn.go

@@ -22,6 +22,12 @@ type interConn struct {
 
 	client bool
 	user   *protocol.MemoryUser
+	lock   chan struct{}
+	mu     sync.Mutex
+}
+
+func (c *interConn) unlock() {
+	<-c.lock
 }
 
 func (c *interConn) User() *protocol.MemoryUser {
@@ -33,6 +39,9 @@ func (c *interConn) Read(b []byte) (int, error) {
 }
 
 func (c *interConn) Write(b []byte) (int, error) {
+	c.lock <- struct{}{}
+	defer c.unlock()
+
 	if c.client {
 		c.client = false
 		if _, err := c.stream.Write(append(quicvarint.Append(nil, FrameTypeTCPRequest), b...)); err != nil {
@@ -40,11 +49,20 @@ func (c *interConn) Write(b []byte) (int, error) {
 		}
 		return len(b), nil
 	}
-
 	return c.stream.Write(b)
 }
 
 func (c *interConn) Close() error {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	select {
+	case c.lock <- struct{}{}:
+		defer c.unlock()
+	default:
+		c.stream.CancelRead(0)
+		c.stream.CancelWrite(0)
+		return nil
+	}
 	c.stream.CancelRead(0)
 	return c.stream.Close()
 }

transport/internet/hysteria/dialer.go

@@ -257,6 +257,7 @@ func (c *client) tcp(ctx context.Context) (stat.Connection, error) {
 		remote: c.conn.RemoteAddr(),
 
 		client: true,
+		lock:   make(chan struct{}, 1),
 	}, nil
 }
 

transport/internet/hysteria/hub.go

@@ -134,6 +134,7 @@ func (h *httpHandler) StreamDispatcher(ft http3.FrameType, stream *quic.Stream,
 			remote: h.conn.RemoteAddr(),
 
 			user: h.user,
+			lock: make(chan struct{}, 1),
 		})
 		return true, nil
 	default: