XHTTP client: Fix a data race in WaitReadCloser (#6694)

https://github.com/XTLS/Xray-core/pull/6694#issuecomment-5575759500

---------

Co-authored-by: 风扇滑翔翼 <Fangliding.fshxy@outlook.com>

common/buf/writer.go

@@ -118,7 +118,9 @@ func (w *BufferedWriter) Write(b []byte) (int, error) {
 
 		nBytes, err := w.buffer.Write(b)
 		totalBytes += nBytes
-		if err != nil {
+
+		// ErrBufferFull means a partial write, so flush below and continue
+		if err != nil && err != ErrBufferFull {
 			return totalBytes, err
 		}
 		if !w.buffered || w.buffer.IsFull() {

go.mod

@@ -32,7 +32,7 @@ require (
 	golang.zx2c4.com/wintun v0.0.0-20230126152724-0fa3db229ce2
 	golang.zx2c4.com/wireguard v0.0.0-20250521234502-f333402bd9cb
 	golang.zx2c4.com/wireguard/windows v1.0.1
-	google.golang.org/grpc v1.83.1
+	google.golang.org/grpc v1.83.2
 	google.golang.org/protobuf v1.36.12
 	gvisor.dev/gvisor v0.0.0-20260122175437-89a5d21be8f0
 	h12.io/socks v1.0.3

go.sum

@@ -160,8 +160,8 @@ gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4=
 gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk=
 google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8=
-google.golang.org/grpc v1.83.1 h1:HIO0+BEtBP6soyqvqC8sNUjZ7bTs+0hFQuFF+RAy++Y=
-google.golang.org/grpc v1.83.1/go.mod h1:kDyl6SKsiHKt0uylY5gtn5cEjkrIOhQOGDgIc4JGwzQ=
+google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU=
+google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8=
 google.golang.org/protobuf v1.36.12 h1:pJOKDDOyeXErUroCihFAd5LQuwXBSpVnKGrj5o/fwxc=
 google.golang.org/protobuf v1.36.12/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
 gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=

infra/conf/blackhole.go

@@ -1,52 +1,42 @@
 package conf
 
 import (
-	"encoding/json"
+	"encoding/base64"
+	"strings"
 
 	"github.com/xtls/xray-core/common/errors"
-	"github.com/xtls/xray-core/common/serial"
 	"github.com/xtls/xray-core/proxy/blackhole"
 	"google.golang.org/protobuf/proto"
 )
 
-type NoneResponse struct{}
-
-func (*NoneResponse) Build() (proto.Message, error) {
-	return new(blackhole.NoneResponse), nil
-}
-
-type HTTPResponse struct{}
-
-func (*HTTPResponse) Build() (proto.Message, error) {
-	return new(blackhole.HTTPResponse), nil
+type ResponseConfig struct {
+	Type               string `json:"type"`
+	CustomResponseData string `json:"customResponseData"`
 }
 
 type BlackholeConfig struct {
-	Response json.RawMessage `json:"response"`
+	Response *ResponseConfig `json:"response"`
 }
 
 func (v *BlackholeConfig) Build() (proto.Message, error) {
 	config := new(blackhole.Config)
 	if v.Response != nil {
-		response, _, err := configLoader.Load(v.Response)
-		if err != nil {
-			return nil, errors.New("Config: Failed to parse Blackhole response config.").Base(err)
+		responseName := strings.ToLower(v.Response.Type)
+		switch responseName {
+		case "none", "":
+			config.Response = &blackhole.Response{Type: "none"}
+		case "http":
+			config.Response = &blackhole.Response{Type: "http"}
+		case "custom":
+			data, err := base64.StdEncoding.DecodeString(v.Response.CustomResponseData)
+			if err != nil {
+				return nil, errors.New("failed to decode custom response data: " + err.Error())
+			}
+			config.Response = &blackhole.Response{Type: "custom", CustomResponseData: data}
+		default:
+			return nil, errors.New("unknown blackhole response: " + responseName)
 		}
-		responseSettings, err := response.(Buildable).Build()
-		if err != nil {
-			return nil, err
-		}
-		config.Response = serial.ToTypedMessage(responseSettings)
 	}
 
 	return config, nil
 }
-
-var configLoader = NewJSONConfigLoader(
-	ConfigCreatorCache{
-		"none": func() interface{} { return new(NoneResponse) },
-		"http": func() interface{} { return new(HTTPResponse) },
-	},
-	"type",
-	"",
-)

infra/conf/blackhole_test.go

@@ -3,7 +3,6 @@ package conf_test
 import (
 	"testing"
 
-	"github.com/xtls/xray-core/common/serial"
 	. "github.com/xtls/xray-core/infra/conf"
 	"github.com/xtls/xray-core/proxy/blackhole"
 )
@@ -22,7 +21,7 @@ func TestHTTPResponseJSON(t *testing.T) {
 			}`,
 			Parser: loadJSON(creator),
 			Output: &blackhole.Config{
-				Response: serial.ToTypedMessage(&blackhole.HTTPResponse{}),
+				Response: &blackhole.Response{Type: "http"},
 			},
 		},
 		{
@@ -32,3 +31,27 @@ func TestHTTPResponseJSON(t *testing.T) {
 		},
 	})
 }
+
+func TestCustomResponseJSON(t *testing.T) {
+	creator := func() Buildable {
+		return new(BlackholeConfig)
+	}
+
+	runMultiTestCase(t, []TestCase{
+		{
+			Input: `{
+				"response": {
+					"type": "custom",
+					"customResponseData": "Y3VzdG9tIHJlc3BvbnNl"
+				}
+			}`,
+			Parser: loadJSON(creator),
+			Output: &blackhole.Config{
+				Response: &blackhole.Response{
+					Type:               "custom",
+					CustomResponseData: []byte("custom response"),
+				},
+			},
+		},
+	})
+}

proxy/blackhole/blackhole.go

@@ -2,12 +2,15 @@
 package blackhole
 
 import (
+	"bytes"
 	"context"
+	"net/http"
 	"time"
 
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
 	"github.com/xtls/xray-core/common/dice"
+	"github.com/xtls/xray-core/common/errors"
 	"github.com/xtls/xray-core/common/net"
 	"github.com/xtls/xray-core/common/session"
 	"github.com/xtls/xray-core/common/signal"
@@ -17,14 +20,34 @@ import (
 
 // Handler is an outbound connection that silently swallow the entire payload.
 type Handler struct {
-	response ResponseConfig
+	response []byte
+}
+
+var http403response = http.Response{
+	StatusCode: 403,
+	ProtoMajor: 1,
+	ProtoMinor: 1,
+	Header: http.Header{
+		"Connection":    {"close"},
+		"Cache-Control": {"max-age=3600, public"},
+	},
 }
 
 // New creates a new blackhole handler.
 func New(ctx context.Context, config *Config) (*Handler, error) {
-	response, err := config.GetInternalResponse()
-	if err != nil {
-		return nil, err
+	response := []byte{}
+	if config.Response != nil {
+		switch config.Response.Type {
+		case "", "none":
+		case "http":
+			var data bytes.Buffer
+			common.Must(http403response.Write(&data))
+			response = data.Bytes()
+		case "custom":
+			response = config.Response.CustomResponseData
+		default:
+			return nil, errors.New("unknown blackhole response: " + config.Response.Type)
+		}
 	}
 	return &Handler{
 		response: response,
@@ -37,8 +60,10 @@ func (h *Handler) Process(ctx context.Context, link *transport.Link, dialer inte
 	ob := outbounds[len(outbounds)-1]
 	ob.Name = "blackhole"
 
-	nBytes := h.response.WriteTo(link.Writer)
-	if nBytes > 0 {
+	if len(h.response) > 0 {
+		mbc := buf.MultiBufferContainer{}
+		common.Must2(mbc.Write(h.response))
+		link.Writer.WriteMultiBuffer(mbc.MultiBuffer)
 		// Sleep a little here to make sure the response is sent to client.
 		time.Sleep(time.Second)
 	}

proxy/blackhole/blackhole_test.go

@@ -1,12 +1,15 @@
 package blackhole_test
 
 import (
+	"bufio"
+	"bytes"
 	"context"
+	"crypto/rand"
+	"net/http"
 	"testing"
 
 	"github.com/xtls/xray-core/common"
 	"github.com/xtls/xray-core/common/buf"
-	"github.com/xtls/xray-core/common/serial"
 	"github.com/xtls/xray-core/common/session"
 	"github.com/xtls/xray-core/proxy/blackhole"
 	"github.com/xtls/xray-core/transport"
@@ -16,27 +19,58 @@ import (
 func TestBlackholeHTTPResponse(t *testing.T) {
 	ctx := session.ContextWithOutbounds(context.Background(), []*session.Outbound{{}})
 	handler, err := blackhole.New(ctx, &blackhole.Config{
-		Response: serial.ToTypedMessage(&blackhole.HTTPResponse{}),
+		Response: &blackhole.Response{Type: "http"},
 	})
 	common.Must(err)
 
 	reader, writer := pipe.New(pipe.WithoutSizeLimit())
 
-	var mb buf.MultiBuffer
-	var rerr error
+	dataCh := make(chan buf.MultiBuffer, 1)
 	go func() {
-		b, e := reader.ReadMultiBuffer()
-		mb = b
-		rerr = e
+		mb := common.Must2(reader.ReadMultiBuffer())
+		dataCh <- mb
 	}()
-
 	link := transport.Link{
 		Reader: reader,
 		Writer: writer,
 	}
 	common.Must(handler.Process(ctx, &link, nil))
-	common.Must(rerr)
-	if mb.IsEmpty() {
-		t.Error("expect http response, but nothing")
+	mb := <-dataCh
+	data := make([]byte, mb.Len())
+	mb.Copy(data)
+	resp := common.Must2(http.ReadResponse(bufio.NewReader(bytes.NewBuffer(data)), nil))
+	if resp.StatusCode != 403 {
+		t.Errorf("expected 403 response, got %d", resp.StatusCode)
+	}
+}
+
+func TestBlackholeCustomResponse(t *testing.T) {
+	ctx := session.ContextWithOutbounds(context.Background(), []*session.Outbound{{}})
+	// slightly bigger than a buffer
+	expected := make([]byte, buf.Size+1000)
+	if _, err := rand.Read(expected); err != nil {
+		t.Fatal(err)
+	}
+	handler, err := blackhole.New(ctx, &blackhole.Config{
+		Response: &blackhole.Response{
+			Type:               "custom",
+			CustomResponseData: expected,
+		},
+	})
+	common.Must(err)
+
+	reader, writer := pipe.New(pipe.WithoutSizeLimit())
+	var actual buf.MultiBuffer
+	var rerr error
+	go func() {
+		actual, rerr = reader.ReadMultiBuffer()
+	}()
+
+	link := transport.Link{Reader: reader, Writer: writer}
+	common.Must(handler.Process(ctx, &link, nil))
+	common.Must(rerr)
+
+	if actual.String() != string(expected) {
+		t.Errorf("custom response mismatch")
 	}
 }

proxy/blackhole/config.go

@@ -1,47 +0,0 @@
-package blackhole
-
-import (
-	"github.com/xtls/xray-core/common"
-	"github.com/xtls/xray-core/common/buf"
-)
-
-const (
-	http403response = `HTTP/1.1 403 Forbidden
-Connection: close
-Cache-Control: max-age=3600, public
-Content-Length: 0
-
-
-`
-)
-
-// ResponseConfig is the configuration for blackhole responses.
-type ResponseConfig interface {
-	// WriteTo writes a predefined response to the specified buffer.
-	WriteTo(buf.Writer) int32
-}
-
-// WriteTo implements ResponseConfig.WriteTo().
-func (*NoneResponse) WriteTo(buf.Writer) int32 { return 0 }
-
-// WriteTo implements ResponseConfig.WriteTo().
-func (*HTTPResponse) WriteTo(writer buf.Writer) int32 {
-	b := buf.New()
-	common.Must2(b.WriteString(http403response))
-	n := b.Len()
-	writer.WriteMultiBuffer(buf.MultiBuffer{b})
-	return n
-}
-
-// GetInternalResponse converts response settings from proto to internal data structure.
-func (c *Config) GetInternalResponse() (ResponseConfig, error) {
-	if c.GetResponse() == nil {
-		return new(NoneResponse), nil
-	}
-
-	config, err := c.GetResponse().GetInstance()
-	if err != nil {
-		return nil, err
-	}
-	return config.(ResponseConfig), nil
-}

proxy/blackhole/config.pb.go

@@ -7,7 +7,6 @@
 package blackhole
 
 import (
-	serial "github.com/xtls/xray-core/common/serial"
 	protoreflect "google.golang.org/protobuf/reflect/protoreflect"
 	protoimpl "google.golang.org/protobuf/runtime/protoimpl"
 	reflect "reflect"
@@ -22,26 +21,28 @@ const (
 	_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
 )
 
-type NoneResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
+type Response struct {
+	state              protoimpl.MessageState `protogen:"open.v1"`
+	Type               string                 `protobuf:"bytes,1,opt,name=type,proto3" json:"type,omitempty"`
+	CustomResponseData []byte                 `protobuf:"bytes,2,opt,name=custom_response_data,json=customResponseData,proto3" json:"custom_response_data,omitempty"`
+	unknownFields      protoimpl.UnknownFields
+	sizeCache          protoimpl.SizeCache
 }
 
-func (x *NoneResponse) Reset() {
-	*x = NoneResponse{}
+func (x *Response) Reset() {
+	*x = Response{}
 	mi := &file_proxy_blackhole_config_proto_msgTypes[0]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
 
-func (x *NoneResponse) String() string {
+func (x *Response) String() string {
 	return protoimpl.X.MessageStringOf(x)
 }
 
-func (*NoneResponse) ProtoMessage() {}
+func (*Response) ProtoMessage() {}
 
-func (x *NoneResponse) ProtoReflect() protoreflect.Message {
+func (x *Response) ProtoReflect() protoreflect.Message {
 	mi := &file_proxy_blackhole_config_proto_msgTypes[0]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
@@ -53,57 +54,35 @@ func (x *NoneResponse) ProtoReflect() protoreflect.Message {
 	return mi.MessageOf(x)
 }
 
-// Deprecated: Use NoneResponse.ProtoReflect.Descriptor instead.
-func (*NoneResponse) Descriptor() ([]byte, []int) {
+// Deprecated: Use Response.ProtoReflect.Descriptor instead.
+func (*Response) Descriptor() ([]byte, []int) {
 	return file_proxy_blackhole_config_proto_rawDescGZIP(), []int{0}
 }
 
-type HTTPResponse struct {
-	state         protoimpl.MessageState `protogen:"open.v1"`
-	unknownFields protoimpl.UnknownFields
-	sizeCache     protoimpl.SizeCache
-}
-
-func (x *HTTPResponse) Reset() {
-	*x = HTTPResponse{}
-	mi := &file_proxy_blackhole_config_proto_msgTypes[1]
-	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-	ms.StoreMessageInfo(mi)
-}
-
-func (x *HTTPResponse) String() string {
-	return protoimpl.X.MessageStringOf(x)
-}
-
-func (*HTTPResponse) ProtoMessage() {}
-
-func (x *HTTPResponse) ProtoReflect() protoreflect.Message {
-	mi := &file_proxy_blackhole_config_proto_msgTypes[1]
+func (x *Response) GetType() string {
 	if x != nil {
-		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
-		if ms.LoadMessageInfo() == nil {
-			ms.StoreMessageInfo(mi)
-		}
-		return ms
+		return x.Type
 	}
-	return mi.MessageOf(x)
+	return ""
 }
 
-// Deprecated: Use HTTPResponse.ProtoReflect.Descriptor instead.
-func (*HTTPResponse) Descriptor() ([]byte, []int) {
-	return file_proxy_blackhole_config_proto_rawDescGZIP(), []int{1}
+func (x *Response) GetCustomResponseData() []byte {
+	if x != nil {
+		return x.CustomResponseData
+	}
+	return nil
 }
 
 type Config struct {
 	state         protoimpl.MessageState `protogen:"open.v1"`
-	Response      *serial.TypedMessage   `protobuf:"bytes,1,opt,name=response,proto3" json:"response,omitempty"`
+	Response      *Response              `protobuf:"bytes,1,opt,name=response,proto3" json:"response,omitempty"`
 	unknownFields protoimpl.UnknownFields
 	sizeCache     protoimpl.SizeCache
 }
 
 func (x *Config) Reset() {
 	*x = Config{}
-	mi := &file_proxy_blackhole_config_proto_msgTypes[2]
+	mi := &file_proxy_blackhole_config_proto_msgTypes[1]
 	ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 	ms.StoreMessageInfo(mi)
 }
@@ -115,7 +94,7 @@ func (x *Config) String() string {
 func (*Config) ProtoMessage() {}
 
 func (x *Config) ProtoReflect() protoreflect.Message {
-	mi := &file_proxy_blackhole_config_proto_msgTypes[2]
+	mi := &file_proxy_blackhole_config_proto_msgTypes[1]
 	if x != nil {
 		ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
 		if ms.LoadMessageInfo() == nil {
@@ -128,10 +107,10 @@ func (x *Config) ProtoReflect() protoreflect.Message {
 
 // Deprecated: Use Config.ProtoReflect.Descriptor instead.
 func (*Config) Descriptor() ([]byte, []int) {
-	return file_proxy_blackhole_config_proto_rawDescGZIP(), []int{2}
+	return file_proxy_blackhole_config_proto_rawDescGZIP(), []int{1}
 }
 
-func (x *Config) GetResponse() *serial.TypedMessage {
+func (x *Config) GetResponse() *Response {
 	if x != nil {
 		return x.Response
 	}
@@ -142,11 +121,12 @@ var File_proxy_blackhole_config_proto protoreflect.FileDescriptor
 
 const file_proxy_blackhole_config_proto_rawDesc = "" +
 	"\n" +
-	"\x1cproxy/blackhole/config.proto\x12\x14xray.proxy.blackhole\x1a!common/serial/typed_message.proto\"\x0e\n" +
-	"\fNoneResponse\"\x0e\n" +
-	"\fHTTPResponse\"F\n" +
-	"\x06Config\x12<\n" +
-	"\bresponse\x18\x01 \x01(\v2 .xray.common.serial.TypedMessageR\bresponseB^\n" +
+	"\x1cproxy/blackhole/config.proto\x12\x14xray.proxy.blackhole\"P\n" +
+	"\bResponse\x12\x12\n" +
+	"\x04type\x18\x01 \x01(\tR\x04type\x120\n" +
+	"\x14custom_response_data\x18\x02 \x01(\fR\x12customResponseData\"D\n" +
+	"\x06Config\x12:\n" +
+	"\bresponse\x18\x01 \x01(\v2\x1e.xray.proxy.blackhole.ResponseR\bresponseB^\n" +
 	"\x18com.xray.proxy.blackholeP\x01Z)github.com/xtls/xray-core/proxy/blackhole\xaa\x02\x14Xray.Proxy.Blackholeb\x06proto3"
 
 var (
@@ -161,15 +141,13 @@ func file_proxy_blackhole_config_proto_rawDescGZIP() []byte {
 	return file_proxy_blackhole_config_proto_rawDescData
 }
 
-var file_proxy_blackhole_config_proto_msgTypes = make([]protoimpl.MessageInfo, 3)
+var file_proxy_blackhole_config_proto_msgTypes = make([]protoimpl.MessageInfo, 2)
 var file_proxy_blackhole_config_proto_goTypes = []any{
-	(*NoneResponse)(nil),        // 0: xray.proxy.blackhole.NoneResponse
-	(*HTTPResponse)(nil),        // 1: xray.proxy.blackhole.HTTPResponse
-	(*Config)(nil),              // 2: xray.proxy.blackhole.Config
-	(*serial.TypedMessage)(nil), // 3: xray.common.serial.TypedMessage
+	(*Response)(nil), // 0: xray.proxy.blackhole.Response
+	(*Config)(nil),   // 1: xray.proxy.blackhole.Config
 }
 var file_proxy_blackhole_config_proto_depIdxs = []int32{
-	3, // 0: xray.proxy.blackhole.Config.response:type_name -> xray.common.serial.TypedMessage
+	0, // 0: xray.proxy.blackhole.Config.response:type_name -> xray.proxy.blackhole.Response
 	1, // [1:1] is the sub-list for method output_type
 	1, // [1:1] is the sub-list for method input_type
 	1, // [1:1] is the sub-list for extension type_name
@@ -188,7 +166,7 @@ func file_proxy_blackhole_config_proto_init() {
 			GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
 			RawDescriptor: unsafe.Slice(unsafe.StringData(file_proxy_blackhole_config_proto_rawDesc), len(file_proxy_blackhole_config_proto_rawDesc)),
 			NumEnums:      0,
-			NumMessages:   3,
+			NumMessages:   2,
 			NumExtensions: 0,
 			NumServices:   0,
 		},

proxy/blackhole/config.proto

@@ -6,12 +6,11 @@ option go_package = "github.com/xtls/xray-core/proxy/blackhole";
 option java_package = "com.xray.proxy.blackhole";
 option java_multiple_files = true;
 
-import "common/serial/typed_message.proto";
-
-message NoneResponse {}
-
-message HTTPResponse {}
+message Response {
+  string type = 1;
+  bytes custom_response_data = 2;
+}
 
 message Config {
-  xray.common.serial.TypedMessage response = 1;
+  Response response = 1;
 }

proxy/blackhole/config_test.go

@@ -1,26 +1,19 @@
 package blackhole_test
 
 import (
-	"bufio"
-	"net/http"
+	"context"
 	"testing"
 
 	"github.com/xtls/xray-core/common"
-	"github.com/xtls/xray-core/common/buf"
-	. "github.com/xtls/xray-core/proxy/blackhole"
+	"github.com/xtls/xray-core/proxy/blackhole"
 )
 
 func TestHTTPResponse(t *testing.T) {
-	buffer := buf.New()
-
-	httpResponse := new(HTTPResponse)
-	httpResponse.WriteTo(buf.NewWriter(buffer))
-
-	reader := bufio.NewReader(buffer)
-	response, err := http.ReadResponse(reader, nil)
+	handler, err := blackhole.New(context.Background(), &blackhole.Config{
+		Response: &blackhole.Response{Type: "http"},
+	})
 	common.Must(err)
-
-	if response.StatusCode != 403 {
-		t.Error("expected status code 403, but got ", response.StatusCode)
+	if handler == nil {
+		t.Error("expected HTTP response handler")
 	}
 }

transport/internet/hysteria/hub.go

@@ -231,8 +231,8 @@ func Listen(ctx context.Context, address net.Address, port net.Port, streamSetti
 				transport.TLSClientConfig.InsecureSkipVerify = true
 			}
 		case "", "unix":
-			u = &url.URL{Scheme: "http", Host: "localhost"}
 			path := u.Path
+			u = &url.URL{Scheme: "http", Host: "localhost"}
 			dialer := &net.Dialer{Timeout: 30 * time.Second}
 			transport = transport.Clone()
 			transport.Proxy = nil

transport/internet/splithttp/client.go

@@ -68,7 +68,7 @@ func (c *DefaultDialerClient) OpenStream(ctx context.Context, url string, sessio
 	}
 	c.transportConfig.FillStreamRequest(req, sessionId, "")
 
-	wrc = &WaitReadCloser{Wait: make(chan struct{})}
+	wrc = &WaitReadCloser{wait: done.New()}
 	go func() {
 		resp, err := c.client.Do(req)
 		if err != nil {
@@ -188,38 +188,35 @@ func (c *DefaultDialerClient) Close() error {
 }
 
 type WaitReadCloser struct {
-	Wait chan struct{}
-	io.ReadCloser
+	wait   *done.Instance
+	reader atomic.Pointer[io.ReadCloser]
 }
 
 func (w *WaitReadCloser) Set(rc io.ReadCloser) {
-	w.ReadCloser = rc
-	defer func() {
-		if recover() != nil {
-			rc.Close()
+	w.reader.Store(&rc)
+	if w.wait.Done() {
+		if p := w.reader.Swap(nil); p != nil {
+			(*p).Close()
 		}
-	}()
-	close(w.Wait)
+	}
+	w.wait.Close()
 }
 
 func (w *WaitReadCloser) Read(b []byte) (int, error) {
-	if w.ReadCloser == nil {
-		if <-w.Wait; w.ReadCloser == nil {
+	rc := w.reader.Load()
+	if rc == nil {
+		<-w.wait.Wait()
+		if rc = w.reader.Load(); rc == nil {
 			return 0, io.ErrClosedPipe
 		}
 	}
-	return w.ReadCloser.Read(b)
+	return (*rc).Read(b)
 }
 
 func (w *WaitReadCloser) Close() error {
-	if w.ReadCloser != nil {
-		return w.ReadCloser.Close()
+	w.wait.Close()
+	if p := w.reader.Swap(nil); p != nil {
+		return (*p).Close()
 	}
-	defer func() {
-		if recover() != nil && w.ReadCloser != nil {
-			w.ReadCloser.Close()
-		}
-	}()
-	close(w.Wait)
 	return nil
 }