From 3d54f9cdc94378130b2e350c0e6fc83e01071979 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Fri, 31 Jul 2026 15:15:44 +0800 Subject: [PATCH 1/8] fix: resolve 7 quality issues (#100-#106) and dialWSS so configured RootCAs actually verify peer certs (MITM fix). return explicit errors instead of nil-deref panics. compression is enabled; wrap snappy writer so every Write is flushed (small packets no longer stuck in buffer). Likely fixes #14. concurrent gc() nil-ing Connection; save & restore write timeout so a per-call timeout does not permanently rewrite the conn deadline; replace CloseConn hard-assert of *tls.Conn with safe type assertions. server.go x2) invisible to go vet due to custom logger wrapper. copy); wire up client.wg so Close() waits for reConnect; synchronize server Listener()/PacketConn()/stop() via RWMutex. (single attempt per call, reConnect owns back-off) so WithReconnectAttempts is honored and unreachable targets no longer spin forever; fix TestTCPClient to dial 127.0.0.1 instead of the unspecified [::] listener address. --- transport/client.go | 219 +++++++++++++++++++++------------------ transport/client_test.go | 6 +- transport/connection.go | 60 +++++++++-- transport/server.go | 49 ++++++--- transport/session.go | 78 +++++++++++--- transport/tls.go | 18 +++- 6 files changed, 285 insertions(+), 145 deletions(-) diff --git a/transport/client.go b/transport/client.go index cb1d712f..d4efd874 100644 --- a/transport/client.go +++ b/transport/client.go @@ -157,29 +157,41 @@ func (c *client) dialTCP() Session { conn net.Conn ) - for { - if c.IsClosed() { - return nil - } - if c.sslEnabled { - if sslConfig, buildTlsConfErr := c.tlsConfigBuilder.BuildTlsConfig(); buildTlsConfErr == nil && sslConfig != nil { - d := &net.Dialer{Timeout: connectTimeout} - conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig) - } + // #106: this function performs a SINGLE dial attempt and returns nil on + // failure. The bounded retry/back-off is owned by reConnect() (which + // honors maxReconnectAttempts). Previously dialTCP had an unbounded + // for{} loop here that never returned on failure, making + // WithReconnectAttempts a no-op and hanging clients/tests forever when + // the target was unreachable. + if c.IsClosed() { + return nil + } + if c.sslEnabled { + // #101: guard against a TLS config builder that returns (nil, nil); + // previously a nil config with nil err fell through and the nil conn + // was dereferenced below (conn.RemoteAddr) -> panic. + sslConfig, buildTlsConfErr := c.tlsConfigBuilder.BuildTlsConfig() + if buildTlsConfErr != nil { + err = buildTlsConfErr + } else if sslConfig == nil { + err = fmt.Errorf("tlsConfigBuilder returned nil config without error") } else { - conn, err = net.DialTimeout("tcp", c.addr, connectTimeout) - } - if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { - _ = conn.Close() - err = errSelfConnect - } - if err == nil { - return newTCPSession(conn, c) + d := &net.Dialer{Timeout: connectTimeout} + conn, err = tls.DialWithDialer(d, "tcp", c.addr, sslConfig) } - - log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err)) - <-gxtime.After(connectInterval) + } else { + conn, err = net.DialTimeout("tcp", c.addr, connectTimeout) + } + if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { + _ = conn.Close() + err = errSelfConnect + } + if err == nil { + return newTCPSession(conn, c) } + + log.Infof("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, connectTimeout, perrors.WithStack(err)) + return nil } func (c *client) dialUDP() Session { @@ -193,51 +205,49 @@ func (c *client) dialUDP() Session { buf []byte ) + // #106: single attempt; reConnect() owns bounded retry/back-off. + if c.IsClosed() { + return nil + } bufp = gxbytes.GetBytes(128) defer gxbytes.PutBytes(bufp) buf = *bufp localAddr = &net.UDPAddr{IP: net.IPv4zero, Port: 0} peerAddr, _ = net.ResolveUDPAddr("udp", c.addr) - for { - if c.IsClosed() { - return nil - } - conn, err = net.DialUDP("udp", localAddr, peerAddr) - if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { - _ = conn.Close() - err = errSelfConnect - } - if err != nil { - log.Warnf("net.DialTimeout(addr:%s, timeout:%v) = error:%+v", c.addr, perrors.WithStack(err)) - <-gxtime.After(connectInterval) - continue - } + conn, err = net.DialUDP("udp", localAddr, peerAddr) + if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { + _ = conn.Close() + err = errSelfConnect + } + if err != nil { + // #104: the format string referenced a timeout verb but no argument + // was supplied; UDP dial has no timeout, so drop the verb. + log.Warnf("net.DialUDP(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) + return nil + } - // check connection alive by write/read action - if err := conn.SetWriteDeadline(time.Now().Add(1e9)); err != nil { - log.Warnf("failed to set write deadline: %+v", err) - } - if length, err = conn.Write(connectPingPackage[:]); err != nil { - _ = conn.Close() - log.Warnf("conn.Write(%s) = {length:%d, err:%+v}", string(connectPingPackage), length, perrors.WithStack(err)) - <-gxtime.After(connectInterval) - continue - } - if err := conn.SetReadDeadline(time.Now().Add(1e9)); err != nil { - log.Warnf("failed to set read deadline: %+v", err) - } - length, err = conn.Read(buf) - if netErr, ok := perrors.Cause(err).(net.Error); ok && netErr.Timeout() { - err = nil - } - if err != nil { - log.Infof("conn{%#v}.Read() = {length:%d, err:%+v}", conn, length, perrors.WithStack(err)) - _ = conn.Close() - <-gxtime.After(connectInterval) - continue - } - return newUDPSession(conn, c) + // check connection alive by write/read action + if err := conn.SetWriteDeadline(time.Now().Add(1e9)); err != nil { + log.Warnf("failed to set write deadline: %+v", err) } + if length, err = conn.Write(connectPingPackage[:]); err != nil { + _ = conn.Close() + log.Warnf("conn.Write(%s) = {length:%d, err:%+v}", string(connectPingPackage), length, perrors.WithStack(err)) + return nil + } + if err := conn.SetReadDeadline(time.Now().Add(1e9)); err != nil { + log.Warnf("failed to set read deadline: %+v", err) + } + length, err = conn.Read(buf) + if netErr, ok := perrors.Cause(err).(net.Error); ok && netErr.Timeout() { + err = nil + } + if err != nil { + log.Infof("conn{%#v}.Read() = {length:%d, err:%+v}", conn, length, perrors.WithStack(err)) + _ = conn.Close() + return nil + } + return newUDPSession(conn, c) } func (c *client) dialWS() Session { @@ -248,29 +258,27 @@ func (c *client) dialWS() Session { ss Session ) + // #106: single attempt; reConnect() owns bounded retry/back-off. + if c.IsClosed() { + return nil + } dialer.EnableCompression = true - for { - if c.IsClosed() { - return nil - } - conn, _, err = dialer.Dial(c.addr, nil) - log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) - if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { - _ = conn.Close() - err = errSelfConnect - } - if err == nil { - ss = newWSSession(conn, c) - if ss.(*session).maxMsgLen > 0 { - conn.SetReadLimit(int64(ss.(*session).maxMsgLen)) - } - - return ss + conn, _, err = dialer.Dial(c.addr, nil) + if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { + _ = conn.Close() + err = errSelfConnect + } + if err == nil { + ss = newWSSession(conn, c) + if ss.(*session).maxMsgLen > 0 { + conn.SetReadLimit(int64(ss.(*session).maxMsgLen)) } - log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) - <-gxtime.After(connectInterval) + return ss } + + log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) + return nil } func (c *client) dialWSS() Session { @@ -285,11 +293,16 @@ func (c *client) dialWSS() Session { ss Session ) + // #106: single attempt; reConnect() owns bounded retry/back-off. + if c.IsClosed() { + return nil + } dialer.EnableCompression = true - config = &tls.Config{ - InsecureSkipVerify: true, - } + // #100: do NOT set InsecureSkipVerify=true here. It disables certificate + // verification entirely and makes the RootCAs configured below useless, + // exposing the WSS client to MITM attacks. + config = &tls.Config{} if c.cert != "" { certPEMBlock, err := os.ReadFile(c.cert) @@ -313,8 +326,9 @@ func (c *client) dialWSS() Session { } certPool = x509.NewCertPool() - for _, c := range config.Certificates { - roots, err = x509.ParseCertificates(c.Certificate[len(c.Certificate)-1]) + // avoid shadowing the receiver `c` with the loop variable. + for _, cert := range config.Certificates { + roots, err = x509.ParseCertificates(cert.Certificate[len(cert.Certificate)-1]) if err != nil { panic(fmt.Sprintf("error parsing server's root cert: %+v\n", perrors.WithStack(err))) } @@ -322,33 +336,28 @@ func (c *client) dialWSS() Session { certPool.AddCert(root) } } - config.InsecureSkipVerify = true + // #100: rely on RootCAs for verification instead of InsecureSkipVerify. config.RootCAs = certPool // dialer.EnableCompression = true dialer.TLSClientConfig = config - for { - if c.IsClosed() { - return nil - } - conn, _, err = dialer.Dial(c.addr, nil) - if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { - _ = conn.Close() - err = errSelfConnect - } - if err == nil { - ss = newWSSession(conn, c) - if ss.(*session).maxMsgLen > 0 { - conn.SetReadLimit(int64(ss.(*session).maxMsgLen)) - } - ss.SetName(defaultWSSSessionName) - - return ss + conn, _, err = dialer.Dial(c.addr, nil) + if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { + _ = conn.Close() + err = errSelfConnect + } + if err == nil { + ss = newWSSession(conn, c) + if ss.(*session).maxMsgLen > 0 { + conn.SetReadLimit(int64(ss.(*session).maxMsgLen)) } + ss.SetName(defaultWSSSessionName) - log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) - <-gxtime.After(connectInterval) + return ss } + + log.Infof("websocket.dialer.Dial(addr:%s) = error:%+v", c.addr, perrors.WithStack(err)) + return nil } func (c *client) dial() Session { @@ -430,6 +439,12 @@ func (c *client) RunEventLoop(newSession NewSessionCallback) { c.Lock() c.newSession = newSession c.Unlock() + // #105: track the reConnect loop so Close()'s wg.Wait() actually waits + // for it to exit. Previously wg was never Add'ed, so Close() returned + // immediately even while reConnect was still running (when RunEventLoop + // is invoked from a goroutine). + c.wg.Add(1) + defer c.wg.Done() c.reConnect() } diff --git a/transport/client_test.go b/transport/client_test.go index 8f982307..dff1d65c 100644 --- a/transport/client_test.go +++ b/transport/client_test.go @@ -93,7 +93,11 @@ func newSessionCallback(session Session, handler *MessageHandler) error { func TestTCPClient(t *testing.T) { listenLocalServer := func() (net.Listener, error) { - listener, err := net.Listen("tcp", ":0") + // #106: bind a concrete loopback address instead of ":0" (which + // resolves to the unspecified "[::]" address). Dialing "[::]:port" + // is not a valid connect destination, so every dial failed and the + // client's reconnect loop hung the test forever. + listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { return nil, err } diff --git a/transport/connection.go b/transport/connection.go index 9f3ff501..efe54566 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -233,6 +233,32 @@ func (t *writeFlusher) Write(p []byte) (int, error) { return n, nil } +// for snappy compress. #102: snappy.NewBufferedWriter buffers writes and only +// emits data on Flush, so small packets would sit in the buffer forever if not +// flushed after every Write. This wrapper flushes on every Write, mirroring the +// flate writeFlusher behavior above. +type snappyWriteFlusher struct { + writer *snappy.Writer + lock sync.Mutex +} + +func newSnappyWriteFlusher(w *snappy.Writer) *snappyWriteFlusher { + return &snappyWriteFlusher{writer: w} +} + +func (s *snappyWriteFlusher) Write(p []byte) (int, error) { + s.lock.Lock() + defer s.lock.Unlock() + n, err := s.writer.Write(p) + if err != nil { + return n, perrors.WithStack(err) + } + if err := s.writer.Flush(); err != nil { + return 0, perrors.WithStack(err) + } + return n, nil +} + // SetCompressType set compress type(tcp: zip/snappy, websocket:zip) func (t *gettyTCPConn) SetCompressType(c CompressType) { switch c { @@ -251,7 +277,9 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) { ioReader := io.Reader(t.conn) t.reader = snappy.NewReader(ioReader) ioWriter := io.Writer(t.conn) - t.writer = snappy.NewBufferedWriter(ioWriter) + // #102: wrap the buffered snappy writer so every Write is flushed, + // otherwise small packets never leave the internal buffer. + t.writer = newSnappyWriteFlusher(snappy.NewBufferedWriter(ioWriter)) default: panic(fmt.Sprintf("illegal comparess type %d", c)) @@ -306,8 +334,23 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) { } if buffers, ok := pkg.([][]byte); ok { - netBuf := net.Buffers(buffers) - lg, err = netBuf.WriteTo(t.conn) + // #102: when compression is enabled the [][]byte path must go through + // t.writer (the compress writer), otherwise it writes raw frames + // directly to t.conn and the peer receives a corrupt mix of + // compressed and uncompressed data. + if t.compress == CompressNone { + netBuf := net.Buffers(buffers) + lg, err = netBuf.WriteTo(t.conn) + } else { + for _, b := range buffers { + var n int + n, err = t.writer.Write(b) + if err != nil { + break + } + lg += int64(n) + } + } if err == nil { t.writeBytes.Add((uint32)(lg)) t.writePkgNum.Add((uint32)(len(buffers))) @@ -338,16 +381,21 @@ func (t *gettyTCPConn) CloseConn(waitSec int) { // } if t.conn != nil { - if writer, ok := t.writer.(*snappy.Writer); ok { - if err := writer.Close(); err != nil { + // #102: snappy writer is now wrapped in *snappyWriteFlusher. + if writer, ok := t.writer.(*snappyWriteFlusher); ok { + if err := writer.writer.Close(); err != nil { log.Errorf("snappy.Writer.Close() = error:%+v", err) } } + // #103: do not hard-assert *tls.Conn; use safe type assertions so a + // non-TLS, non-TCP conn does not panic here. if conn, ok := t.conn.(*net.TCPConn); ok { _ = conn.SetLinger(waitSec) _ = conn.Close() + } else if tlsConn, ok := t.conn.(*tls.Conn); ok { + _ = tlsConn.Close() } else { - _ = t.conn.(*tls.Conn).Close() + _ = t.conn.Close() } t.conn = nil } diff --git a/transport/server.go b/transport/server.go index 625e80e6..cea44e95 100644 --- a/transport/server.go +++ b/transport/server.go @@ -81,7 +81,7 @@ type server struct { // net pktListener net.PacketConn streamListener net.Listener - lock sync.Mutex // for server + lock sync.RWMutex // for server endPointType EndPointType server *http.Server // for ws or wss server sync.Once @@ -160,15 +160,20 @@ func (s *server) stop() { cancel() } s.server = nil + // #105: read & nil the listeners under s.lock so concurrent + // Listener()/PacketConn() accessors don't race with stop(). + streamListener := s.streamListener + s.streamListener = nil + pktListener := s.pktListener + s.pktListener = nil s.lock.Unlock() - if s.streamListener != nil { + // close outside the lock to avoid blocking other lock holders. + if streamListener != nil { // let the server exit asap when got error from RunEventLoop. - _ = s.streamListener.Close() - s.streamListener = nil + _ = streamListener.Close() } - if s.pktListener != nil { - _ = s.pktListener.Close() - s.pktListener = nil + if pktListener != nil { + _ = pktListener.Close() } }) } @@ -202,13 +207,21 @@ func (s *server) listenTCP() error { return perrors.Wrapf(err, "gxnet.ListenOnTCPRandomPort(addr:%s)", s.addr) } } else { - if s.sslEnabled { - if sslConfig, buildTlsConfErr := s.tlsConfigBuilder.BuildTlsConfig(); buildTlsConfErr == nil && sslConfig != nil { - streamListener, err = tls.Listen("tcp", s.addr, sslConfig) - } - } else { - streamListener, err = net.Listen("tcp", s.addr) + if s.sslEnabled { + // #101: guard against a TLS config builder that returns (nil, nil); + // previously a nil config with nil err fell through, leaving + // streamListener nil, and s.streamListener.Addr() panicked below. + sslConfig, buildTlsConfErr := s.tlsConfigBuilder.BuildTlsConfig() + if buildTlsConfErr != nil { + return perrors.Wrapf(buildTlsConfErr, "BuildTlsConfig(addr:%s)", s.addr) + } + if sslConfig == nil { + return fmt.Errorf("BuildTlsConfig returned nil config without error for addr:%s", s.addr) } + streamListener, err = tls.Listen("tcp", s.addr, sslConfig) + } else { + streamListener, err = net.Listen("tcp", s.addr) + } if err != nil { return perrors.Wrapf(err, "net.Listen(tcp, addr:%s)", s.addr) } @@ -267,7 +280,7 @@ func (s *server) accept(newSession NewSessionCallback) (Session, error) { return nil, perrors.WithStack(err) } if gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { - log.Warnf("conn.localAddr{%s} == conn.RemoteAddr", conn.LocalAddr().String(), conn.RemoteAddr().String()) + log.Warnf("conn.localAddr{%s} == conn.RemoteAddr{%s}", conn.LocalAddr().String(), conn.RemoteAddr().String()) return nil, perrors.WithStack(errSelfConnect) } @@ -384,7 +397,7 @@ func (s *wsHandler) serveWSRequest(w http.ResponseWriter, r *http.Request) { } if conn.RemoteAddr().String() == conn.LocalAddr().String() { _ = conn.Close() - log.Warnf("conn.localAddr{%s} == conn.RemoteAddr", conn.LocalAddr().String(), conn.RemoteAddr().String()) + log.Warnf("conn.localAddr{%s} == conn.RemoteAddr{%s}", conn.LocalAddr().String(), conn.RemoteAddr().String()) return } // conn.SetReadLimit(int64(handler.maxMsgLen)) @@ -513,10 +526,16 @@ func (s *server) RunEventLoop(newSession NewSessionCallback) { } func (s *server) Listener() net.Listener { + // #105: guard against concurrent stop() which nils s.streamListener. + s.lock.RLock() + defer s.lock.RUnlock() return s.streamListener } func (s *server) PacketConn() net.PacketConn { + // #105: guard against concurrent stop() which nils s.pktListener. + s.lock.RLock() + defer s.lock.RUnlock() return s.pktListener } diff --git a/transport/session.go b/transport/session.go index 9d0d9903..7a2e140e 100644 --- a/transport/session.go +++ b/transport/session.go @@ -193,14 +193,26 @@ func newWSSession(conn *websocket.Conn, endPoint EndPoint) Session { } func (s *session) Reset() { - *s = session{ - name: defaultSessionName, - once: &sync.Once{}, - done: make(chan struct{}), - period: period, - wait: pendingDuration, - attrs: gxcontext.NewValuesContext(context.Background()), - } + // #105: Reset() previously did `*s = session{...}` without holding s.lock, + // racing with concurrent readers. Reset the fields individually under the + // lock instead; replacing the whole struct would also copy the mutexes + // (flagged by `go vet`). + s.lock.Lock() + defer s.lock.Unlock() + s.name = defaultSessionName + s.endPoint = nil + s.Connection = nil + s.listener = nil + s.reader = nil + s.writer = nil + s.maxMsgLen = 0 + s.once = &sync.Once{} + s.done = make(chan struct{}) + s.period = period + s.wait = pendingDuration + s.attrs = gxcontext.NewValuesContext(context.Background()) + s.grNum.Store(0) + s.closeCallback = callbacks{} } func (s *session) Conn() net.Conn { @@ -416,10 +428,28 @@ func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, s } s.packetLock.RLock() defer s.packetLock.RUnlock() + // #103: read s.Connection under s.lock to guard against concurrent gc() + // which sets s.Connection = nil; a bare s.Connection.Send below could + // otherwise nil-deref. The obtained conn/gc pointers stay valid even if + // gc() later nils the field, because they reference the underlying obj. + s.lock.RLock() + conn := s.Connection + gc := s.gettyConn() + s.lock.RUnlock() + if conn == nil || gc == nil { + return 0, 0, ErrSessionClosed + } + var origWriteTimeout time.Duration + if 0 < timeout { + // #103: save & restore so a per-call timeout does not permanently + // rewrite the connection's write deadline for subsequent writes. + origWriteTimeout = gc.WriteTimeout() + gc.SetWriteTimeout(timeout) + } + successCount, err = conn.Send(pkg) if 0 < timeout { - s.gettyConn().SetWriteTimeout(timeout) + gc.SetWriteTimeout(origWriteTimeout) } - successCount, err = s.Connection.Send(pkg) if err != nil { log.Warnf("%s, [session.WritePkg] @s.Connection.Write(pkg:%#v) = err:%+v", s.Stat(), pkg, err) return len(pkgBytes), successCount, perrors.WithStack(err) @@ -442,8 +472,16 @@ func (s *session) WriteBytes(pkg []byte) (int, error) { defer s.packetLock.RUnlock() } + // #103: guard s.Connection against concurrent gc() nil-ing it. + s.lock.RLock() + conn := s.Connection + s.lock.RUnlock() + if conn == nil { + return 0, ErrSessionClosed + } + for leftPackageSize > maxPacketLen { - _, err := s.Connection.Send(pkg[writeSize:(writeSize + maxPacketLen)]) + _, err := conn.Send(pkg[writeSize:(writeSize + maxPacketLen)]) if err != nil { return writeSize, perrors.Wrapf(err, "s.Connection.Write(pkg len:%d)", len(pkg)) } @@ -455,7 +493,7 @@ func (s *session) WriteBytes(pkg []byte) (int, error) { return writeSize, nil } - _, err := s.Connection.Send(pkg[writeSize:]) + _, err := conn.Send(pkg[writeSize:]) if err != nil { return writeSize, perrors.Wrapf(err, "s.Connection.Write(pkg len:%d)", len(pkg)) } @@ -473,10 +511,17 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) (int, error) { } // reduce syscall and memcopy for multiple packages - if _, ok := s.Connection.(*gettyTCPConn); ok { + // #103: guard s.Connection against concurrent gc() nil-ing it. + s.lock.RLock() + conn := s.Connection + s.lock.RUnlock() + if conn == nil { + return 0, ErrSessionClosed + } + if _, ok := conn.(*gettyTCPConn); ok { s.packetLock.RLock() defer s.packetLock.RUnlock() - lg, err := s.Connection.Send(pkgs) + lg, err := conn.Send(pkgs) if err != nil { return 0, perrors.Wrapf(err, "s.Connection.Write(pkgs num:%d)", len(pkgs)) } @@ -691,8 +736,9 @@ func (s *session) handleTCPPackage() error { // as https://github.com/apache/dubbo-getty/issues/77#issuecomment-939652203 // this branch is impossible. Even if it happens, the bufLen will be zero and the error // is io.EOF when getty continues to read the socket. - exit = false - log.Infof("%s, session.conn read EOF, while the bufLen(%d) is non-zero.", s.sessionToken()) + exit = false + // #104: missing bufLen argument for the %d verb. + log.Infof("%s, session.conn read EOF, while the bufLen(%d) is non-zero.", s.sessionToken(), bufLen) } break } diff --git a/transport/tls.go b/transport/tls.go index 86323eb3..c137d332 100644 --- a/transport/tls.go +++ b/transport/tls.go @@ -74,7 +74,10 @@ func (s *ServerTlsConfigBuilder) BuildTlsConfig() (*tls.Config, error) { certPool = x509.NewCertPool() if ok := certPool.AppendCertsFromPEM(certPem); !ok { log.Error("failed to parse root certificate file") - return nil, err + // #101: err is nil here (from a successful os.ReadFile above), + // returning (nil, nil) causes a nil-pointer panic at the caller + // (listenTCP). Return an explicit error instead. + return nil, fmt.Errorf("failed to parse root certificate file: %s", s.ServerTrustCertCollectionPath) } config.ClientCAs = certPool config.ClientAuth = tls.RequireAnyClientCert @@ -107,11 +110,16 @@ func (c *ClientTlsConfigBuilder) BuildTlsConfig() (*tls.Config, error) { ok := clientCertPool.AppendCertsFromPEM(certBytes) if !ok { log.Error("failed to parse root certificate") - return nil, err + // #101: err is nil here (from a successful os.ReadFile above), + // returning (nil, nil) causes a nil-pointer panic at the caller + // (dialTCP/dialWSS). Return an explicit error instead. + return nil, fmt.Errorf("failed to parse root certificate: %s", c.ClientTrustCertCollectionPath) } return &tls.Config{ - RootCAs: clientCertPool, - Certificates: []tls.Certificate{cert}, - InsecureSkipVerify: true, + RootCAs: clientCertPool, + Certificates: []tls.Certificate{cert}, + // #100: do NOT set InsecureSkipVerify=true here; it disables + // certificate verification entirely and makes the RootCAs configured + // above useless, exposing the client to MITM attacks. }, nil } From a538d94c5637aaa02b5bee963ee830e3afcd68e6 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Fri, 31 Jul 2026 16:06:33 +0800 Subject: [PATCH 2/8] style: normalize line endings to LF for CI fmt check From eb9480181d7a2c37d86bab561254c1bfd1588d72 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Fri, 31 Jul 2026 16:47:09 +0800 Subject: [PATCH 3/8] Harden reconnect and session concurrency Serialize per-write timeout mutation, track every reconnect loop during client shutdown, and protect callback reset with its owning mutex. Add deterministic regression coverage for retry bounds, cancellation, lifecycle tracking, timeout restoration, and callback synchronization. Constraint: Preserve existing public APIs and keep the change within transport lifecycle paths. Confidence: High; targeted regressions pass repeatedly and static and build checks pass. Scope-risk: Reconnect scheduling, client shutdown, callback reset, and timed session writes. Tested: Focused regressions x10; related non-network transport tests; go vet ./transport ./util; go build ./transport ./util; gofmt and imports-formatter. Not-tested: Full Windows WSS suite and Linux race execution; the pushed PR CI will provide Ubuntu coverage. Co-authored-by: OmX --- transport/client.go | 94 ++++++++++++--------- transport/client_test.go | 127 +++++++++++++++++++++++++++++ transport/server.go | 28 +++---- transport/session.go | 26 +++--- transport/session_callback_test.go | 51 ++++++++++++ transport/session_test.go | 92 +++++++++++++++++++++ 6 files changed, 356 insertions(+), 62 deletions(-) diff --git a/transport/client.go b/transport/client.go index d4efd874..fa3e2001 100644 --- a/transport/client.go +++ b/transport/client.go @@ -22,7 +22,6 @@ import ( "crypto/x509" "encoding/pem" "fmt" - "math" "net" "os" "strings" @@ -439,21 +438,36 @@ func (c *client) RunEventLoop(newSession NewSessionCallback) { c.Lock() c.newSession = newSession c.Unlock() - // #105: track the reConnect loop so Close()'s wg.Wait() actually waits - // for it to exit. Previously wg was never Add'ed, so Close() returned - // immediately even while reConnect was still running (when RunEventLoop - // is invoked from a goroutine). - c.wg.Add(1) - defer c.wg.Done() - c.reConnect() + <-c.runReconnect() +} + +// runReconnect starts a reconnect loop only while the client is open. The +// client lock serializes WaitGroup.Add with stop, which prevents Add racing +// with Close's Wait. +func (c *client) runReconnect() <-chan struct{} { + done := make(chan struct{}) + c.Lock() + select { + case <-c.done: + c.Unlock() + close(done) + return done + default: + c.wg.Add(1) + } + c.Unlock() + + go func() { + defer c.wg.Done() + defer close(done) + c.reConnect() + }() + return done } // a for-loop connect to make sure the connection pool is valid func (c *client) reConnect() { - var ( - sessionNum, reconnectAttempts int - maxReconnectInterval int64 - ) + var reconnectAttempts int reconnectInterval := c.reconnectInterval if reconnectInterval == 0 { reconnectInterval = defaultReconnectInterval @@ -463,42 +477,48 @@ func (c *client) reConnect() { maxReconnectAttempts = defaultMaxReconnectAttempts } connPoolSize := c.number - for { + for reconnectAttempts < maxReconnectAttempts { if c.IsClosed() { log.Warnf("client{peer:%s} goroutine exit now.", c.addr) - break + return } - sessionNum = c.sessionNum() - if connPoolSize <= sessionNum || maxReconnectAttempts < reconnectAttempts { - //exit reconnect when the number of connection pools is sufficient or the current reconnection attempts exceeds the max reconnection attempts. - break + if connPoolSize <= c.sessionNum() { + return } c.connect() reconnectAttempts++ - maxReconnectInterval = int64(math.Min(float64(reconnectAttempts), float64(maxBackOffTimes))) * int64(reconnectInterval) - <-gxtime.After(time.Duration(maxReconnectInterval)) + if c.IsClosed() || connPoolSize <= c.sessionNum() || reconnectAttempts >= maxReconnectAttempts { + return + } + + backOffTimes := reconnectAttempts + if maxBackOffTimes < backOffTimes { + backOffTimes = maxBackOffTimes + } + backoff := time.Duration(int64(backOffTimes) * int64(reconnectInterval)) + select { + case <-c.done: + return + case <-gxtime.After(backoff): + } } } func (c *client) stop() { - select { - case <-c.done: - return - default: - c.Do(func() { - close(c.done) - c.Lock() - for s := range c.ssMap { - s.RemoveAttribute(sessionClientKey) - s.RemoveAttribute(ignoreReconnectKey) - s.Close() - } - c.ssMap = nil - - c.Unlock() - }) - } + c.Do(func() { + c.Lock() + close(c.done) + sessions := c.ssMap + c.ssMap = nil + c.Unlock() + + for s := range sessions { + s.RemoveAttribute(sessionClientKey) + s.RemoveAttribute(ignoreReconnectKey) + s.Close() + } + }) } func (c *client) IsClosed() bool { diff --git a/transport/client_test.go b/transport/client_test.go index dff1d65c..7c0bfa18 100644 --- a/transport/client_test.go +++ b/transport/client_test.go @@ -19,11 +19,14 @@ package getty import ( "bytes" + "crypto/tls" + "errors" "net" "net/http" "os" "strconv" "sync" + "sync/atomic" "testing" "time" ) @@ -36,6 +39,130 @@ import ( type PackageHandler struct{} +var errTestTLSConfig = errors.New("test TLS config failure") + +type countingTLSConfigBuilder struct { + calls atomic.Int32 + entered chan struct{} +} + +func (b *countingTLSConfigBuilder) BuildTlsConfig() (*tls.Config, error) { + b.calls.Add(1) + select { + case b.entered <- struct{}{}: + default: + } + return nil, errTestTLSConfig +} + +func newFailingReconnectClient(builder TlsConfigBuilder, interval time.Duration, attempts int) *client { + return newClient(TCP_CLIENT, + WithServerAddress("127.0.0.1:1"), + WithConnectionNumber(1), + WithReconnectInterval(int(interval)), + WithReconnectAttempts(attempts), + WithClientSslEnabled(true), + WithClientTlsConfigBuilder(builder), + ) +} + +func TestReconnectAttemptsAreExactAndSkipFinalBackoff(t *testing.T) { + builder := &countingTLSConfigBuilder{entered: make(chan struct{}, 4)} + clt := newFailingReconnectClient(builder, 100*time.Millisecond, 3) + + started := time.Now() + clt.RunEventLoop(func(Session) error { return nil }) + elapsed := time.Since(started) + + if got := builder.calls.Load(); got != 3 { + t.Fatalf("TLS config build calls = %d, want exactly 3 reconnect attempts", got) + } + // Correct behavior waits after attempts one and two only: 100ms + 200ms. + // A final backoff adds another 300ms. + if elapsed >= 500*time.Millisecond { + t.Fatalf("reconnect loop took %v; it appears to wait after the final attempt", elapsed) + } +} + +func TestReconnectBackoffIsCancelledByClose(t *testing.T) { + builder := &countingTLSConfigBuilder{entered: make(chan struct{}, 4)} + clt := newFailingReconnectClient(builder, 2*time.Second, 3) + + eventLoopDone := make(chan struct{}) + go func() { + clt.RunEventLoop(func(Session) error { return nil }) + close(eventLoopDone) + }() + select { + case <-builder.entered: + case <-time.After(time.Second): + t.Fatal("first reconnect attempt did not start") + } + time.Sleep(20 * time.Millisecond) + + closeDone := make(chan struct{}) + go func() { + clt.Close() + close(closeDone) + }() + + timedOut := false + select { + case <-closeDone: + case <-time.After(200 * time.Millisecond): + timedOut = true + } + if timedOut { + select { + case <-closeDone: + case <-time.After(3 * time.Second): + t.Fatal("Close did not return after the current backoff elapsed") + } + t.Fatal("Close did not cancel the reconnect backoff") + } + select { + case <-eventLoopDone: + case <-time.After(time.Second): + t.Fatal("RunEventLoop did not return after Close") + } +} + +func TestSessionReconnectIsTrackedByClose(t *testing.T) { + builder := &countingTLSConfigBuilder{entered: make(chan struct{}, 4)} + clt := newFailingReconnectClient(builder, 2*time.Second, 3) + localConn, peerConn := net.Pipe() + defer func() { + _ = localConn.Close() + _ = peerConn.Close() + }() + ss := newTCPSession(localConn, clt).(*session) + ss.SetAttribute(sessionClientKey, clt) + ss.SetAttribute(ignoreReconnectKey, false) + + sessionStopDone := make(chan struct{}) + go func() { + ss.stop() + close(sessionStopDone) + }() + select { + case <-builder.entered: + case <-time.After(time.Second): + t.Fatal("session-triggered reconnect did not start") + } + + clt.Close() + select { + case <-sessionStopDone: + case <-time.After(100 * time.Millisecond): + select { + case <-sessionStopDone: + case <-time.After(3 * time.Second): + t.Fatal("session stop did not return after reconnect backoff elapsed") + } + t.Fatal("Close returned while the session-triggered reconnect was still running") + } +} + func (h *PackageHandler) Read(ss Session, data []byte) (any, int, error) { return nil, 0, nil } diff --git a/transport/server.go b/transport/server.go index cea44e95..f26da611 100644 --- a/transport/server.go +++ b/transport/server.go @@ -207,21 +207,21 @@ func (s *server) listenTCP() error { return perrors.Wrapf(err, "gxnet.ListenOnTCPRandomPort(addr:%s)", s.addr) } } else { - if s.sslEnabled { - // #101: guard against a TLS config builder that returns (nil, nil); - // previously a nil config with nil err fell through, leaving - // streamListener nil, and s.streamListener.Addr() panicked below. - sslConfig, buildTlsConfErr := s.tlsConfigBuilder.BuildTlsConfig() - if buildTlsConfErr != nil { - return perrors.Wrapf(buildTlsConfErr, "BuildTlsConfig(addr:%s)", s.addr) - } - if sslConfig == nil { - return fmt.Errorf("BuildTlsConfig returned nil config without error for addr:%s", s.addr) + if s.sslEnabled { + // #101: guard against a TLS config builder that returns (nil, nil); + // previously a nil config with nil err fell through, leaving + // streamListener nil, and s.streamListener.Addr() panicked below. + sslConfig, buildTlsConfErr := s.tlsConfigBuilder.BuildTlsConfig() + if buildTlsConfErr != nil { + return perrors.Wrapf(buildTlsConfErr, "BuildTlsConfig(addr:%s)", s.addr) + } + if sslConfig == nil { + return fmt.Errorf("BuildTlsConfig returned nil config without error for addr:%s", s.addr) + } + streamListener, err = tls.Listen("tcp", s.addr, sslConfig) + } else { + streamListener, err = net.Listen("tcp", s.addr) } - streamListener, err = tls.Listen("tcp", s.addr, sslConfig) - } else { - streamListener, err = net.Listen("tcp", s.addr) - } if err != nil { return perrors.Wrapf(err, "net.Listen(tcp, addr:%s)", s.addr) } diff --git a/transport/session.go b/transport/session.go index 7a2e140e..956cfb33 100644 --- a/transport/session.go +++ b/transport/session.go @@ -197,6 +197,8 @@ func (s *session) Reset() { // racing with concurrent readers. Reset the fields individually under the // lock instead; replacing the whole struct would also copy the mutexes // (flagged by `go vet`). + s.closeCallbackMutex.Lock() + defer s.closeCallbackMutex.Unlock() s.lock.Lock() defer s.lock.Unlock() s.name = defaultSessionName @@ -426,8 +428,13 @@ func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, s } else { pkg = pkgBytes } - s.packetLock.RLock() - defer s.packetLock.RUnlock() + if 0 < timeout { + s.packetLock.Lock() + defer s.packetLock.Unlock() + } else { + s.packetLock.RLock() + defer s.packetLock.RUnlock() + } // #103: read s.Connection under s.lock to guard against concurrent gc() // which sets s.Connection = nil; a bare s.Connection.Send below could // otherwise nil-deref. The obtained conn/gc pointers stay valid even if @@ -439,17 +446,14 @@ func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, s if conn == nil || gc == nil { return 0, 0, ErrSessionClosed } - var origWriteTimeout time.Duration if 0 < timeout { // #103: save & restore so a per-call timeout does not permanently // rewrite the connection's write deadline for subsequent writes. - origWriteTimeout = gc.WriteTimeout() + origWriteTimeout := gc.WriteTimeout() gc.SetWriteTimeout(timeout) + defer gc.SetWriteTimeout(origWriteTimeout) } successCount, err = conn.Send(pkg) - if 0 < timeout { - gc.SetWriteTimeout(origWriteTimeout) - } if err != nil { log.Warnf("%s, [session.WritePkg] @s.Connection.Write(pkg:%#v) = err:%+v", s.Stat(), pkg, err) return len(pkgBytes), successCount, perrors.WithStack(err) @@ -736,9 +740,9 @@ func (s *session) handleTCPPackage() error { // as https://github.com/apache/dubbo-getty/issues/77#issuecomment-939652203 // this branch is impossible. Even if it happens, the bufLen will be zero and the error // is io.EOF when getty continues to read the socket. - exit = false - // #104: missing bufLen argument for the %d verb. - log.Infof("%s, session.conn read EOF, while the bufLen(%d) is non-zero.", s.sessionToken(), bufLen) + exit = false + // #104: missing bufLen argument for the %d verb. + log.Infof("%s, session.conn read EOF, while the bufLen(%d) is non-zero.", s.sessionToken(), bufLen) } break } @@ -934,7 +938,7 @@ func (s *session) stop() { clt, cltFound := s.GetAttribute(sessionClientKey).(*client) ignoreReconnect, flagFound := s.GetAttribute(ignoreReconnectKey).(bool) if cltFound && flagFound && !ignoreReconnect { - clt.reConnect() + clt.runReconnect() } }) } diff --git a/transport/session_callback_test.go b/transport/session_callback_test.go index 68228fb2..1bb538fa 100644 --- a/transport/session_callback_test.go +++ b/transport/session_callback_test.go @@ -237,3 +237,54 @@ func TestSessionCallback(t *testing.T) { } }) } + +func TestResetWaitsForCloseCallbacks(t *testing.T) { + s := &session{ + once: &sync.Once{}, + done: make(chan struct{}), + closeCallback: callbacks{}, + } + + callbackStarted := make(chan struct{}) + releaseCallback := make(chan struct{}) + s.AddCloseCallback("test", "blocking", func() { + close(callbackStarted) + <-releaseCallback + }) + + callbackDone := make(chan struct{}) + go func() { + s.invokeCloseCallbacks() + close(callbackDone) + }() + <-callbackStarted + + resetDone := make(chan struct{}) + go func() { + s.Reset() + close(resetDone) + }() + + select { + case <-resetDone: + close(releaseCallback) + <-callbackDone + t.Fatal("Reset returned while a close callback was still running") + case <-time.After(50 * time.Millisecond): + } + + close(releaseCallback) + select { + case <-callbackDone: + case <-time.After(time.Second): + t.Fatal("close callback did not finish") + } + select { + case <-resetDone: + case <-time.After(time.Second): + t.Fatal("Reset did not finish after the close callback completed") + } + if got := s.closeCallback.Len(); got != 0 { + t.Fatalf("callback count after Reset = %d, want 0", got) + } +} diff --git a/transport/session_test.go b/transport/session_test.go index bbbc9a26..e62c48b2 100644 --- a/transport/session_test.go +++ b/transport/session_test.go @@ -19,6 +19,7 @@ package getty import ( "errors" + "io" "net" "testing" "time" @@ -32,6 +33,97 @@ func (errorReader) Read(Session, []byte) (any, int, error) { return nil, 0, errTestReadFailure } +type timeoutTestWriter struct{} + +func (timeoutTestWriter) Write(Session, any) ([]byte, error) { + return []byte("x"), nil +} + +type timeoutTestCall struct { + observed time.Duration + release chan struct{} +} + +type timeoutTestNetConn struct { + owner *gettyTCPConn + entered chan *timeoutTestCall +} + +func (c *timeoutTestNetConn) Write(p []byte) (int, error) { + call := &timeoutTestCall{ + observed: c.owner.WriteTimeout(), + release: make(chan struct{}), + } + c.entered <- call + <-call.release + return len(p), nil +} + +func (*timeoutTestNetConn) Read([]byte) (int, error) { return 0, io.EOF } +func (*timeoutTestNetConn) Close() error { return nil } +func (*timeoutTestNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } +func (*timeoutTestNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } +func (*timeoutTestNetConn) SetDeadline(time.Time) error { return nil } +func (*timeoutTestNetConn) SetReadDeadline(time.Time) error { return nil } +func (*timeoutTestNetConn) SetWriteDeadline(time.Time) error { return nil } + +func TestConcurrentWritePkgTimeoutRestoration(t *testing.T) { + netConn := &timeoutTestNetConn{entered: make(chan *timeoutTestCall, 2)} + ss := newTCPSession(netConn, nil).(*session) + ss.writer = timeoutTestWriter{} + conn := ss.Connection.(*gettyTCPConn) + netConn.owner = conn + initialTimeout := conn.WriteTimeout() + + firstDone := make(chan error, 1) + go func() { + _, _, err := ss.WritePkg("first", 3*time.Second) + firstDone <- err + }() + firstCall := <-netConn.entered + + secondDone := make(chan error, 1) + go func() { + _, _, err := ss.WritePkg("second", 5*time.Second) + secondDone <- err + }() + + select { + case secondCall := <-netConn.entered: + close(firstCall.release) + <-firstDone + close(secondCall.release) + <-secondDone + t.Fatal("second timed write entered while the first still owned the shared write timeout") + case <-time.After(50 * time.Millisecond): + } + + if firstCall.observed != 3*time.Second { + t.Fatalf("first write observed timeout %v, want %v", firstCall.observed, 3*time.Second) + } + close(firstCall.release) + if err := <-firstDone; err != nil { + t.Fatalf("first write failed: %v", err) + } + + var secondCall *timeoutTestCall + select { + case secondCall = <-netConn.entered: + case <-time.After(time.Second): + t.Fatal("second timed write did not enter after the first completed") + } + if secondCall.observed != 5*time.Second { + t.Fatalf("second write observed timeout %v, want %v", secondCall.observed, 5*time.Second) + } + close(secondCall.release) + if err := <-secondDone; err != nil { + t.Fatalf("second write failed: %v", err) + } + if got := conn.WriteTimeout(); got != initialTimeout { + t.Fatalf("write timeout after concurrent calls = %v, want %v", got, initialTimeout) + } +} + func TestHandlePackageWithNilListenerDoesNotPanicOnError(t *testing.T) { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { From a299dc0a25d41eba03cd00aca1f0067b8681c9d3 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 1 Aug 2026 06:26:15 +0800 Subject: [PATCH 4/8] Close remaining transport lifecycle races Synchronize session reset with package, callback, heartbeat, and connection cleanup work so pooled sessions cannot be cleared while old-generation activity is still using them. Preserve published server listeners across shutdown, serialize snappy close with writes, process final EOF bytes once, and build WSS trust roots without panics. Constraint: Preserve existing reconnect behavior and use system trust roots when no custom WSS CA is configured. Confidence: High; deterministic regression tests and repeated WSL race runs cover the changed paths. Scope-risk: Medium; session lifecycle and TLS trust construction affect core teardown and dial paths. Tested: WSL Go 1.25 go test -race ./transport ./util; WSL go test ./...; targeted race tests at count 10; TestUDPClient at count 3; go vet ./transport ./util; go build ./transport ./util; golangci-lint run ./... --timeout=10m; gofmt/import formatting; git diff --check. Not-tested: Windows full transport loopback suite remains blocked by the previously reproduced local loopback failure; GitHub CI will validate the pushed Head. Co-authored-by: OmX --- transport/client.go | 75 +++++------- transport/client_test.go | 53 +++++++++ transport/connection.go | 12 +- transport/connection_test.go | 128 +++++++++++++++++++++ transport/server.go | 18 +-- transport/server_test.go | 36 ++++++ transport/session.go | 217 ++++++++++++++++++++++++++++++----- transport/session_test.go | 202 +++++++++++++++++++++++++++++++- transport/tls.go | 1 + transport/tls_test.go | 53 +++++++++ 10 files changed, 707 insertions(+), 88 deletions(-) create mode 100644 transport/connection_test.go create mode 100644 transport/tls_test.go diff --git a/transport/client.go b/transport/client.go index fa3e2001..47ad3671 100644 --- a/transport/client.go +++ b/transport/client.go @@ -20,7 +20,6 @@ package getty import ( "crypto/tls" "crypto/x509" - "encoding/pem" "fmt" "net" "os" @@ -280,16 +279,31 @@ func (c *client) dialWS() Session { return nil } +func (c *client) buildWSSClientTLSConfig() (*tls.Config, error) { + config := &tls.Config{MinVersion: tls.VersionTLS12} + if c.cert == "" { + return config, nil + } + + certPEM, err := os.ReadFile(c.cert) + if err != nil { + return nil, perrors.Wrapf(err, "os.ReadFile(cert:%s)", c.cert) + } + certPool := x509.NewCertPool() + if !certPool.AppendCertsFromPEM(certPEM) { + return nil, fmt.Errorf("failed to parse root certificate: %s", c.cert) + } + config.RootCAs = certPool + return config, nil +} + func (c *client) dialWSS() Session { var ( - err error - root *x509.Certificate - roots []*x509.Certificate - certPool *x509.CertPool - config *tls.Config - dialer websocket.Dialer - conn *websocket.Conn - ss Session + err error + config *tls.Config + dialer websocket.Dialer + conn *websocket.Conn + ss Session ) // #106: single attempt; reConnect() owns bounded retry/back-off. @@ -298,47 +312,12 @@ func (c *client) dialWSS() Session { } dialer.EnableCompression = true - // #100: do NOT set InsecureSkipVerify=true here. It disables certificate - // verification entirely and makes the RootCAs configured below useless, - // exposing the WSS client to MITM attacks. - config = &tls.Config{} - - if c.cert != "" { - certPEMBlock, err := os.ReadFile(c.cert) - if err != nil { - panic(fmt.Sprintf("os.ReadFile(cert:%s) = error:%+v", c.cert, perrors.WithStack(err))) - } - - var cert tls.Certificate - for { - var certDERBlock *pem.Block - certDERBlock, certPEMBlock = pem.Decode(certPEMBlock) - if certDERBlock == nil { - break - } - if certDERBlock.Type == "CERTIFICATE" { - cert.Certificate = append(cert.Certificate, certDERBlock.Bytes) - } - } - config.Certificates = make([]tls.Certificate, 1) - config.Certificates[0] = cert - } - - certPool = x509.NewCertPool() - // avoid shadowing the receiver `c` with the loop variable. - for _, cert := range config.Certificates { - roots, err = x509.ParseCertificates(cert.Certificate[len(cert.Certificate)-1]) - if err != nil { - panic(fmt.Sprintf("error parsing server's root cert: %+v\n", perrors.WithStack(err))) - } - for _, root = range roots { - certPool.AddCert(root) - } + config, err = c.buildWSSClientTLSConfig() + if err != nil { + log.Errorf("build WSS client TLS config = error:%+v", perrors.WithStack(err)) + return nil } - // #100: rely on RootCAs for verification instead of InsecureSkipVerify. - config.RootCAs = certPool - // dialer.EnableCompression = true dialer.TLSClientConfig = config conn, _, err = dialer.Dial(c.addr, nil) if err == nil && gxnet.IsSameAddr(conn.RemoteAddr(), conn.LocalAddr()) { diff --git a/transport/client_test.go b/transport/client_test.go index 7c0bfa18..0a44a8d0 100644 --- a/transport/client_test.go +++ b/transport/client_test.go @@ -24,6 +24,7 @@ import ( "net" "net/http" "os" + "path/filepath" "strconv" "sync" "sync/atomic" @@ -590,6 +591,58 @@ func DownloadFile(filepath string, content []byte) error { return err } +func TestBuildWSSClientTLSConfig(t *testing.T) { + t.Run("system roots", func(t *testing.T) { + config, err := (&client{}).buildWSSClientTLSConfig() + if err != nil { + t.Fatal(err) + } + if config.RootCAs != nil { + t.Fatal("RootCAs must be nil when no custom root certificate is configured") + } + if config.MinVersion != tls.VersionTLS12 { + t.Fatalf("MinVersion = %d, want TLS 1.2 (%d)", config.MinVersion, tls.VersionTLS12) + } + }) + + t.Run("custom root certificate", func(t *testing.T) { + certPath := filepath.Join(t.TempDir(), "root.crt") + if err := os.WriteFile(certPath, WssClientCRT, 0o600); err != nil { + t.Fatal(err) + } + config, err := (&client{ClientOptions: ClientOptions{cert: certPath}}).buildWSSClientTLSConfig() + if err != nil { + t.Fatal(err) + } + if config.RootCAs == nil { + t.Fatal("RootCAs is nil with a configured root certificate") + } + if len(config.Certificates) != 0 { + t.Fatalf("client Certificates contains %d entries, want 0 for a root-only option", len(config.Certificates)) + } + if config.MinVersion != tls.VersionTLS12 { + t.Fatalf("MinVersion = %d, want TLS 1.2 (%d)", config.MinVersion, tls.VersionTLS12) + } + }) + + t.Run("invalid PEM", func(t *testing.T) { + certPath := filepath.Join(t.TempDir(), "invalid.crt") + if err := os.WriteFile(certPath, []byte("not a certificate"), 0o600); err != nil { + t.Fatal(err) + } + if _, err := (&client{ClientOptions: ClientOptions{cert: certPath}}).buildWSSClientTLSConfig(); err == nil { + t.Fatal("invalid PEM returned nil error") + } + }) + + t.Run("missing file", func(t *testing.T) { + certPath := filepath.Join(t.TempDir(), "missing.crt") + if _, err := (&client{ClientOptions: ClientOptions{cert: certPath}}).buildWSSClientTLSConfig(); err == nil { + t.Fatal("missing root certificate returned nil error") + } + }) +} + func TestNewWSSClient(t *testing.T) { var ( err error diff --git a/transport/connection.go b/transport/connection.go index efe54566..091fe18b 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -127,7 +127,7 @@ func (c *gettyConn) GetActive() time.Time { // removed unused methods send/close -func (c gettyConn) ReadTimeout() time.Duration { +func (c *gettyConn) ReadTimeout() time.Duration { return c.rTimeout.Load() } @@ -150,7 +150,7 @@ func (c *gettyConn) SetReadTimeout(rTimeout time.Duration) { } } -func (c gettyConn) WriteTimeout() time.Duration { +func (c *gettyConn) WriteTimeout() time.Duration { return c.wTimeout.Load() } @@ -259,6 +259,12 @@ func (s *snappyWriteFlusher) Write(p []byte) (int, error) { return n, nil } +func (s *snappyWriteFlusher) Close() error { + s.lock.Lock() + defer s.lock.Unlock() + return perrors.WithStack(s.writer.Close()) +} + // SetCompressType set compress type(tcp: zip/snappy, websocket:zip) func (t *gettyTCPConn) SetCompressType(c CompressType) { switch c { @@ -383,7 +389,7 @@ func (t *gettyTCPConn) CloseConn(waitSec int) { if t.conn != nil { // #102: snappy writer is now wrapped in *snappyWriteFlusher. if writer, ok := t.writer.(*snappyWriteFlusher); ok { - if err := writer.writer.Close(); err != nil { + if err := writer.Close(); err != nil { log.Errorf("snappy.Writer.Close() = error:%+v", err) } } diff --git a/transport/connection_test.go b/transport/connection_test.go new file mode 100644 index 00000000..dfb9aa15 --- /dev/null +++ b/transport/connection_test.go @@ -0,0 +1,128 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package getty + +import ( + "io" + "net" + "sync" + "testing" + "time" +) + +import ( + "github.com/golang/snappy" +) + +type blockingSnappyWriter struct { + entered chan struct{} + release chan struct{} +} + +func (w *blockingSnappyWriter) Write(p []byte) (int, error) { + select { + case w.entered <- struct{}{}: + default: + } + <-w.release + return len(p), nil +} + +type timeoutAccessorNetConn struct{} + +func (*timeoutAccessorNetConn) Read([]byte) (int, error) { return 0, io.EOF } +func (*timeoutAccessorNetConn) Write(p []byte) (int, error) { return len(p), nil } +func (*timeoutAccessorNetConn) Close() error { return nil } +func (*timeoutAccessorNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } +func (*timeoutAccessorNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } +func (*timeoutAccessorNetConn) SetDeadline(time.Time) error { return nil } +func (*timeoutAccessorNetConn) SetReadDeadline(time.Time) error { return nil } +func (*timeoutAccessorNetConn) SetWriteDeadline(time.Time) error { return nil } + +func TestConnectionTimeoutAccessorsDoNotCopyAtomicState(t *testing.T) { + conn := newGettyTCPConn(&timeoutAccessorNetConn{}) + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + + go func() { + defer wg.Done() + <-start + for i := 0; i < 10000; i++ { + conn.rLastDeadline.Store(time.Unix(0, int64(i))) + conn.wLastDeadline.Store(time.Unix(0, int64(i))) + } + }() + go func() { + defer wg.Done() + <-start + for i := 0; i < 10000; i++ { + _ = conn.ReadTimeout() + _ = conn.WriteTimeout() + } + }() + + close(start) + wg.Wait() +} + +func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) { + underlying := &blockingSnappyWriter{ + entered: make(chan struct{}, 1), + release: make(chan struct{}), + } + defer func() { + select { + case <-underlying.release: + default: + close(underlying.release) + } + }() + + writer := newSnappyWriteFlusher(snappy.NewBufferedWriter(underlying)) + writeDone := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("payload")) + writeDone <- err + }() + + select { + case <-underlying.entered: + case <-time.After(time.Second): + t.Fatal("snappy write did not reach the underlying writer") + } + + closeDone := make(chan error, 1) + go func() { + closeDone <- writer.Close() + }() + + select { + case err := <-closeDone: + t.Fatalf("Close returned before the active Write completed: %v", err) + case <-time.After(50 * time.Millisecond): + } + + close(underlying.release) + if err := <-writeDone; err != nil { + t.Fatalf("Write failed: %v", err) + } + if err := <-closeDone; err != nil { + t.Fatalf("Close failed: %v", err) + } +} diff --git a/transport/server.go b/transport/server.go index f26da611..3008462c 100644 --- a/transport/server.go +++ b/transport/server.go @@ -160,12 +160,10 @@ func (s *server) stop() { cancel() } s.server = nil - // #105: read & nil the listeners under s.lock so concurrent - // Listener()/PacketConn() accessors don't race with stop(). + // Snapshot the listeners under s.lock. Keep the published listener + // objects in place so event loops cannot race with a nil assignment. streamListener := s.streamListener - s.streamListener = nil pktListener := s.pktListener - s.pktListener = nil s.lock.Unlock() // close outside the lock to avoid blocking other lock holders. if streamListener != nil { @@ -227,8 +225,11 @@ func (s *server) listenTCP() error { } } + addr := streamListener.Addr().String() + s.lock.Lock() s.streamListener = streamListener - s.addr = s.streamListener.Addr().String() + s.addr = addr + s.lock.Unlock() return nil } @@ -256,8 +257,11 @@ func (s *server) listenUDP() error { } } + addr := pktListener.LocalAddr().String() + s.lock.Lock() s.pktListener = pktListener - s.addr = s.pktListener.LocalAddr().String() + s.addr = addr + s.lock.Unlock() return nil } @@ -526,14 +530,12 @@ func (s *server) RunEventLoop(newSession NewSessionCallback) { } func (s *server) Listener() net.Listener { - // #105: guard against concurrent stop() which nils s.streamListener. s.lock.RLock() defer s.lock.RUnlock() return s.streamListener } func (s *server) PacketConn() net.PacketConn { - // #105: guard against concurrent stop() which nils s.pktListener. s.lock.RLock() defer s.lock.RUnlock() return s.pktListener diff --git a/transport/server_test.go b/transport/server_test.go index b3fb689c..c9a94ef9 100644 --- a/transport/server_test.go +++ b/transport/server_test.go @@ -177,6 +177,42 @@ func testUDPServer(t *testing.T, address string) { t.Logf("@address:%s, udp server addr: %v", address, addr) } +func TestServerCloseKeepsPublishedListener(t *testing.T) { + t.Run("TCP", func(t *testing.T) { + server := newServer(TCP_SERVER, WithLocalAddress("127.0.0.1:0")) + if err := server.listen(); err != nil { + t.Fatal(err) + } + listener := server.Listener() + if listener == nil { + t.Fatal("listen did not publish the TCP listener") + } + + server.Close() + + if got := server.Listener(); got != listener { + t.Fatalf("Listener() after Close = %v, want the published listener %v", got, listener) + } + }) + + t.Run("UDP", func(t *testing.T) { + server := newServer(UDP_ENDPOINT, WithLocalAddress("127.0.0.1:0")) + if err := server.listen(); err != nil { + t.Fatal(err) + } + listener := server.PacketConn() + if listener == nil { + t.Fatal("listen did not publish the UDP listener") + } + + server.Close() + + if got := server.PacketConn(); got != listener { + t.Fatalf("PacketConn() after Close = %v, want the published listener %v", got, listener) + } + }) +} + func TestServer(t *testing.T) { var addr string diff --git a/transport/session.go b/transport/session.go index 956cfb33..ff1634f0 100644 --- a/transport/session.go +++ b/transport/session.go @@ -124,7 +124,9 @@ type session struct { maxMsgLen int32 // heartbeat - period time.Duration + period time.Duration + heartbeatTimer *gxtime.Timer + lifecycle *sessionLifecycle // done wait time.Duration @@ -136,6 +138,7 @@ type session struct { // goroutines sync grNum uatomic.Int32 + grWG sync.WaitGroup lock sync.RWMutex packetLock sync.RWMutex @@ -144,6 +147,51 @@ type session struct { closeCallbackMutex sync.RWMutex } +type sessionLifecycle struct { + lock sync.Mutex + closed bool + wg sync.WaitGroup +} + +func newSessionLifecycle() *sessionLifecycle { + return &sessionLifecycle{} +} + +func (l *sessionLifecycle) acquire() bool { + l.lock.Lock() + defer l.lock.Unlock() + if l.closed { + return false + } + l.wg.Add(1) + return true +} + +func (l *sessionLifecycle) release() { + l.wg.Done() +} + +func (l *sessionLifecycle) addClosingTask() { + l.lock.Lock() + l.wg.Add(1) + l.lock.Unlock() +} + +func (l *sessionLifecycle) close() { + l.lock.Lock() + l.closed = true + l.lock.Unlock() +} + +func (l *sessionLifecycle) wait() { + l.wg.Wait() +} + +type heartbeatContext struct { + session *session + lifecycle *sessionLifecycle +} + func newSession(endPoint EndPoint, conn Connection) *session { ss := &session{ name: defaultSessionName, @@ -153,7 +201,8 @@ func newSession(endPoint EndPoint, conn Connection) *session { maxMsgLen: maxReadBufLen, - period: period, + period: period, + lifecycle: newSessionLifecycle(), once: &sync.Once{}, done: make(chan struct{}), @@ -193,6 +242,11 @@ func newWSSession(conn *websocket.Conn, endPoint EndPoint) Session { } func (s *session) Reset() { + lifecycle := s.stopHeartbeat() + s.grWG.Wait() + if lifecycle != nil { + lifecycle.wait() + } // #105: Reset() previously did `*s = session{...}` without holding s.lock, // racing with concurrent readers. Reset the fields individually under the // lock instead; replacing the whole struct would also copy the mutexes @@ -211,6 +265,8 @@ func (s *session) Reset() { s.once = &sync.Once{} s.done = make(chan struct{}) s.period = period + s.heartbeatTimer = nil + s.lifecycle = newSessionLifecycle() s.wait = pendingDuration s.attrs = gxcontext.NewValuesContext(context.Background()) s.grNum.Store(0) @@ -234,6 +290,8 @@ func (s *session) Conn() net.Conn { } func (s *session) EndPoint() EndPoint { + s.lock.RLock() + defer s.lock.RUnlock() return s.endPoint } @@ -271,8 +329,11 @@ func (s *session) Stat() string { // IsClosed check whether the session has been closed. func (s *session) IsClosed() bool { + s.lock.RLock() + done := s.done + s.lock.RUnlock() select { - case <-s.done: + case <-done: return true default: @@ -385,12 +446,24 @@ func (s *session) RemoveAttribute(key any) { } func (s *session) sessionToken() string { - if s.IsClosed() || s.Connection == nil { + s.lock.RLock() + done := s.done + conn := s.Connection + name := s.name + endPoint := s.endPoint + s.lock.RUnlock() + + select { + case <-done: + return "session-closed" + default: + } + if conn == nil || endPoint == nil { return "session-closed" } return fmt.Sprintf("{%s:%s:%d:%s<->%s}", - s.name, s.EndPoint().EndPointType(), s.ID(), s.LocalAddr(), s.RemoteAddr()) + name, endPoint.EndPointType(), conn.ID(), conn.LocalAddr(), conn.RemoteAddr()) } func (s *session) WritePkg(pkg any, timeout time.Duration) (pkgBytesLenth int, successCount int, err error) { @@ -571,13 +644,27 @@ func (s *session) WriteBytesArray(pkgs ...[]byte) (int, error) { } func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error { - ss, _ := arg.(*session) - if ss == nil || ss.IsClosed() { + ctx, _ := arg.(*heartbeatContext) + if ctx == nil || ctx.session == nil || ctx.lifecycle == nil { return ErrSessionClosed } + ss := ctx.session f := func() { - wsConn, wsFlag := ss.Connection.(*gettyWSConn) + if !ctx.lifecycle.acquire() { + return + } + defer ctx.lifecycle.release() + + ss.lock.RLock() + conn := ss.Connection + listener := ss.listener + ss.lock.RUnlock() + if conn == nil || listener == nil { + return + } + + wsConn, wsFlag := conn.(*gettyWSConn) if wsFlag { err := wsConn.writePing() if err != nil { @@ -585,11 +672,17 @@ func heartbeat(_ gxtime.TimerID, _ time.Time, arg any) error { } } - ss.listener.OnCron(ss) + listener.OnCron(ss) } // if enable task pool, run @f asynchronously. - if taskPool := ss.EndPoint().GetTaskPool(); taskPool != nil { + ss.lock.RLock() + endPoint := ss.endPoint + ss.lock.RUnlock() + if endPoint == nil { + return ErrSessionClosed + } + if taskPool := endPoint.GetTaskPool(); taskPool != nil { taskPool.AddTaskAlways(f) return nil } @@ -614,28 +707,62 @@ func (s *session) run() { return } - if _, err := defaultTimerWheel.AddTimer(heartbeat, gxtime.TimerLoop, s.period, s); err != nil { + s.lock.Lock() + lifecycle := s.lifecycle + timer, err := defaultTimerWheel.AddTimer( + heartbeat, + gxtime.TimerLoop, + s.period, + &heartbeatContext{session: s, lifecycle: lifecycle}, + ) + if err == nil { + s.heartbeatTimer = timer + } + s.lock.Unlock() + if err != nil { panic(fmt.Sprintf("failed to add session %s to defaultTimerWheel err:%v", s.Stat(), err)) } s.grNum.Add(1) + s.grWG.Add(1) // start read gr - go s.handlePackage() + go func() { + defer s.grWG.Done() + s.handlePackage() + }() } func (s *session) addTask(pkg any) { + s.lock.RLock() + lifecycle := s.lifecycle + endPoint := s.endPoint + s.lock.RUnlock() + f := func() { + if lifecycle == nil || !lifecycle.acquire() { + return + } + defer lifecycle.release() + // If the session is closed, there is no need to perform CPU-intensive operations. if s.IsClosed() { - log.Errorf("[Id:%d, name=%s, endpoint=%s] Session is closed", s.ID(), s.name, s.EndPoint()) + log.Errorf("%s Session is closed", s.sessionToken()) + return + } + s.lock.RLock() + listener := s.listener + s.lock.RUnlock() + if listener == nil { return } - s.listener.OnMessage(s, pkg) + listener.OnMessage(s, pkg) s.IncReadPkgNum() } - if taskPool := s.EndPoint().GetTaskPool(); taskPool != nil { - taskPool.AddTaskAlways(f) - return + if endPoint != nil { + if taskPool := endPoint.GetTaskPool(); taskPool != nil { + taskPool.AddTaskAlways(f) + return + } } f() } @@ -737,11 +864,8 @@ func (s *session) handleTCPPackage() error { err = nil exit = true if bufLen != 0 { - // as https://github.com/apache/dubbo-getty/issues/77#issuecomment-939652203 - // this branch is impossible. Even if it happens, the bufLen will be zero and the error - // is io.EOF when getty continues to read the socket. - exit = false - // #104: missing bufLen argument for the %d verb. + // Process the bytes returned with EOF below, then exit + // without issuing another read on the closed stream. log.Infof("%s, session.conn read EOF, while the bufLen(%d) is non-zero.", s.sessionToken(), bufLen) } break @@ -919,8 +1043,19 @@ func (s *session) stop() { } } close(s.done) + lifecycle := s.stopHeartbeat() - go func(sessionToken string) { + s.closeCallbackMutex.RLock() + closeCallbacks := s.closeCallback + s.closeCallbackMutex.RUnlock() + if lifecycle != nil { + lifecycle.addClosingTask() + } + + go func(sessionToken string, closeCallbacks callbacks, lifecycle *sessionLifecycle) { + if lifecycle != nil { + defer lifecycle.release() + } defer func() { if r := recover(); r != nil { const size = 64 << 10 @@ -932,8 +1067,8 @@ func (s *session) stop() { } }() - s.invokeCloseCallbacks() - }(s.sessionToken()) + closeCallbacks.Invoke() + }(s.sessionToken(), closeCallbacks, lifecycle) clt, cltFound := s.GetAttribute(sessionClientKey).(*client) ignoreReconnect, flagFound := s.GetAttribute(ignoreReconnectKey).(bool) @@ -944,22 +1079,48 @@ func (s *session) stop() { } } +func (s *session) stopHeartbeat() *sessionLifecycle { + s.lock.Lock() + timer := s.heartbeatTimer + s.heartbeatTimer = nil + lifecycle := s.lifecycle + s.lock.Unlock() + + if timer != nil { + timer.Stop() + } + if lifecycle != nil { + lifecycle.close() + } + return lifecycle +} + func (s *session) gc() { var conn Connection + var wait time.Duration + var lifecycle *sessionLifecycle s.lock.Lock() if s.attrs != nil { s.attrs = nil conn = s.Connection s.Connection = nil + wait = s.wait + lifecycle = s.lifecycle } s.lock.Unlock() + if conn != nil && lifecycle != nil { + lifecycle.addClosingTask() + } - go func() { + go func(conn Connection, wait time.Duration, lifecycle *sessionLifecycle) { + if conn != nil && lifecycle != nil { + defer lifecycle.release() + } if conn != nil { - conn.CloseConn(int(s.wait)) + conn.CloseConn(int(wait)) } - }() + }(conn, wait, lifecycle) } // Close will be invoked by NewSessionCallback(if return error is not nil) diff --git a/transport/session_test.go b/transport/session_test.go index e62c48b2..645c85a7 100644 --- a/transport/session_test.go +++ b/transport/session_test.go @@ -21,11 +21,16 @@ import ( "errors" "io" "net" + "sync" + "sync/atomic" "testing" "time" ) -var errTestReadFailure = errors.New("test read failure") +var ( + errTestReadFailure = errors.New("test read failure") + errUnexpectedSecondRead = errors.New("unexpected second read") +) type errorReader struct{} @@ -67,6 +72,87 @@ func (*timeoutTestNetConn) SetDeadline(time.Time) error { return nil } func (*timeoutTestNetConn) SetReadDeadline(time.Time) error { return nil } func (*timeoutTestNetConn) SetWriteDeadline(time.Time) error { return nil } +type eofDataNetConn struct { + data []byte + reads int +} + +func (c *eofDataNetConn) Read(p []byte) (int, error) { + c.reads++ + if c.reads == 1 { + return copy(p, c.data), io.EOF + } + return 0, errUnexpectedSecondRead +} + +func (*eofDataNetConn) Write(p []byte) (int, error) { return len(p), nil } +func (*eofDataNetConn) Close() error { return nil } +func (*eofDataNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } +func (*eofDataNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } +func (*eofDataNetConn) SetDeadline(time.Time) error { return nil } +func (*eofDataNetConn) SetReadDeadline(time.Time) error { return nil } +func (*eofDataNetConn) SetWriteDeadline(time.Time) error { return nil } + +type wholeFrameReader struct{} + +func (wholeFrameReader) Read(_ Session, data []byte) (any, int, error) { + return string(data), len(data), nil +} + +type recordingEventListener struct { + messages []any +} + +func (*recordingEventListener) OnOpen(Session) error { return nil } +func (*recordingEventListener) OnClose(Session) {} +func (*recordingEventListener) OnError(Session, error) {} +func (*recordingEventListener) OnCron(Session) {} +func (l *recordingEventListener) OnMessage(_ Session, v any) { + l.messages = append(l.messages, v) +} + +type blockingCronEventListener struct { + entered chan struct{} + release chan struct{} + enteredOnce sync.Once +} + +func (*blockingCronEventListener) OnOpen(Session) error { return nil } +func (*blockingCronEventListener) OnClose(Session) {} +func (*blockingCronEventListener) OnError(Session, error) {} +func (l *blockingCronEventListener) OnCron(Session) { + l.enteredOnce.Do(func() { close(l.entered) }) + <-l.release +} +func (*blockingCronEventListener) OnMessage(Session, any) {} + +type resetBarrierNetConn struct { + entered chan struct{} + release chan struct{} + enteredOnce sync.Once + releaseOnce sync.Once + deadlines atomic.Int32 +} + +func (c *resetBarrierNetConn) Read([]byte) (int, error) { + c.enteredOnce.Do(func() { close(c.entered) }) + <-c.release + return 0, io.EOF +} + +func (*resetBarrierNetConn) Write(p []byte) (int, error) { return len(p), nil } +func (*resetBarrierNetConn) Close() error { return nil } +func (*resetBarrierNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } +func (*resetBarrierNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } +func (*resetBarrierNetConn) SetDeadline(time.Time) error { return nil } +func (c *resetBarrierNetConn) SetReadDeadline(time.Time) error { + if c.deadlines.Add(1) > 1 { + c.releaseOnce.Do(func() { close(c.release) }) + } + return nil +} +func (*resetBarrierNetConn) SetWriteDeadline(time.Time) error { return nil } + func TestConcurrentWritePkgTimeoutRestoration(t *testing.T) { netConn := &timeoutTestNetConn{entered: make(chan *timeoutTestCall, 2)} ss := newTCPSession(netConn, nil).(*session) @@ -124,6 +210,120 @@ func TestConcurrentWritePkgTimeoutRestoration(t *testing.T) { } } +func TestResetWaitsForPackageLoop(t *testing.T) { + netConn := &resetBarrierNetConn{ + entered: make(chan struct{}), + release: make(chan struct{}), + } + defer netConn.releaseOnce.Do(func() { close(netConn.release) }) + + ss := newTCPSession(netConn, newServer(TCP_SERVER)).(*session) + ss.SetReader(wholeFrameReader{}) + ss.SetWriter(timeoutTestWriter{}) + ss.SetEventListener(&recordingEventListener{}) + ss.run() + + select { + case <-netConn.entered: + case <-time.After(time.Second): + t.Fatal("package loop did not enter Read") + } + + resetDone := make(chan struct{}) + go func() { + ss.Reset() + close(resetDone) + }() + + select { + case <-resetDone: + t.Fatal("Reset returned while the package loop was still running") + case <-time.After(50 * time.Millisecond): + } + + ss.Close() + select { + case <-resetDone: + case <-time.After(time.Second): + t.Fatal("Reset did not return after Close released the package loop") + } + if ss.Connection != nil { + t.Fatal("Reset did not clear the session connection") + } + if got := ss.grNum.Load(); got != 0 { + t.Fatalf("goroutine count after Reset = %d, want 0", got) + } +} + +func TestResetWaitsForActiveHeartbeat(t *testing.T) { + listener := &blockingCronEventListener{ + entered: make(chan struct{}), + release: make(chan struct{}), + } + ss := newTCPSession(&eofDataNetConn{}, newServer(TCP_SERVER)).(*session) + ss.SetEventListener(listener) + lifecycle := ss.lifecycle + ctx := &heartbeatContext{session: ss, lifecycle: lifecycle} + + heartbeatDone := make(chan error, 1) + go func() { + heartbeatDone <- heartbeat(0, time.Time{}, ctx) + }() + select { + case <-listener.entered: + case <-time.After(time.Second): + t.Fatal("heartbeat did not enter OnCron") + } + + resetDone := make(chan struct{}) + go func() { + ss.Reset() + close(resetDone) + }() + select { + case <-resetDone: + t.Fatal("Reset returned while the heartbeat callback was still running") + case <-time.After(50 * time.Millisecond): + } + + close(listener.release) + select { + case err := <-heartbeatDone: + if err != nil { + t.Fatalf("heartbeat returned error: %v", err) + } + case <-time.After(time.Second): + t.Fatal("heartbeat did not return") + } + select { + case <-resetDone: + case <-time.After(time.Second): + t.Fatal("Reset did not return after the heartbeat callback completed") + } + + if err := heartbeat(0, time.Time{}, ctx); !errors.Is(err, ErrSessionClosed) { + t.Fatalf("stale heartbeat returned %v, want %v", err, ErrSessionClosed) + } +} + +func TestHandleTCPPackageProcessesDataAndStopsOnEOF(t *testing.T) { + netConn := &eofDataNetConn{data: []byte("final frame")} + ss := newTCPSession(netConn, newServer(TCP_SERVER)).(*session) + listener := &recordingEventListener{} + ss.SetReader(wholeFrameReader{}) + ss.SetEventListener(listener) + + if err := ss.handleTCPPackage(); err != nil { + t.Fatalf("handleTCPPackage returned error: %v", err) + } + if netConn.reads != 1 { + t.Fatalf("underlying Read calls = %d, want 1", netConn.reads) + } + if len(listener.messages) != 1 || listener.messages[0] != "final frame" { + t.Fatalf("delivered messages = %#v, want [\"final frame\"]", listener.messages) + } +} + func TestHandlePackageWithNilListenerDoesNotPanicOnError(t *testing.T) { listener, err := net.Listen("tcp", "127.0.0.1:0") if err != nil { diff --git a/transport/tls.go b/transport/tls.go index c137d332..6c37473d 100644 --- a/transport/tls.go +++ b/transport/tls.go @@ -116,6 +116,7 @@ func (c *ClientTlsConfigBuilder) BuildTlsConfig() (*tls.Config, error) { return nil, fmt.Errorf("failed to parse root certificate: %s", c.ClientTrustCertCollectionPath) } return &tls.Config{ + MinVersion: tls.VersionTLS12, RootCAs: clientCertPool, Certificates: []tls.Certificate{cert}, // #100: do NOT set InsecureSkipVerify=true here; it disables diff --git a/transport/tls_test.go b/transport/tls_test.go new file mode 100644 index 00000000..485b34e0 --- /dev/null +++ b/transport/tls_test.go @@ -0,0 +1,53 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package getty + +import ( + "crypto/tls" + "os" + "path/filepath" + "testing" +) + +func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { + tempDir := t.TempDir() + certPath := filepath.Join(tempDir, "client.crt") + keyPath := filepath.Join(tempDir, "client.key") + caPath := filepath.Join(tempDir, "ca.crt") + for path, data := range map[string][]byte{ + certPath: WssServerCRT, + keyPath: WssServerKEY, + caPath: WssClientCRT, + } { + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + } + + config, err := (&ClientTlsConfigBuilder{ + ClientKeyCertChainPath: certPath, + ClientPrivateKeyPath: keyPath, + ClientTrustCertCollectionPath: caPath, + }).BuildTlsConfig() + if err != nil { + t.Fatal(err) + } + if config.MinVersion != tls.VersionTLS12 { + t.Fatalf("MinVersion = %d, want TLS 1.2 (%d)", config.MinVersion, tls.VersionTLS12) + } +} From 26d44cdd56d8d05fdfd140939ac5a7e85cd0dbf2 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 1 Aug 2026 07:00:00 +0800 Subject: [PATCH 5/8] Prevent listener publication after shutdown Make listener publication and server shutdown share one synchronization boundary so a listener created concurrently with Close cannot escape after the server is already closed. Harden the related concurrency and TLS regression tests to assert lifecycle ordering and invalid trust input explicitly. Constraint: Preserve the existing listener accessor behavior after a listener has been published and avoid widening public server lifecycle semantics. Confidence: High; deterministic TCP and UDP close-versus-publish tests pass together with WSL race, repository tests, vet, and build checks. Scope-risk: Limited to transport server shutdown synchronization and review-requested regression test strengthening. Tested: WSL Go 1.25 targeted go test -race; go test ./...; go vet ./transport ./util; go build ./transport ./util; git diff --cached --check. Not-tested: GitHub CI and CodeRabbit on this new commit are pending. Co-authored-by: OmX --- transport/client_test.go | 38 ++++++++++++---- transport/connection_test.go | 12 ++--- transport/server.go | 32 ++++++++++++++ transport/server_test.go | 85 ++++++++++++++++++++++++++++++++++++ transport/session_test.go | 24 +++++----- transport/tls_test.go | 37 ++++++++++++++++ 6 files changed, 198 insertions(+), 30 deletions(-) diff --git a/transport/client_test.go b/transport/client_test.go index 0a44a8d0..5b8afd95 100644 --- a/transport/client_test.go +++ b/transport/client_test.go @@ -45,6 +45,7 @@ var errTestTLSConfig = errors.New("test TLS config failure") type countingTLSConfigBuilder struct { calls atomic.Int32 entered chan struct{} + release <-chan struct{} } func (b *countingTLSConfigBuilder) BuildTlsConfig() (*tls.Config, error) { @@ -53,6 +54,9 @@ func (b *countingTLSConfigBuilder) BuildTlsConfig() (*tls.Config, error) { case b.entered <- struct{}{}: default: } + if b.release != nil { + <-b.release + } return nil, errTestTLSConfig } @@ -129,7 +133,11 @@ func TestReconnectBackoffIsCancelledByClose(t *testing.T) { } func TestSessionReconnectIsTrackedByClose(t *testing.T) { - builder := &countingTLSConfigBuilder{entered: make(chan struct{}, 4)} + releaseReconnect := make(chan struct{}) + builder := &countingTLSConfigBuilder{ + entered: make(chan struct{}, 4), + release: releaseReconnect, + } clt := newFailingReconnectClient(builder, 2*time.Second, 3) localConn, peerConn := net.Pipe() defer func() { @@ -151,16 +159,28 @@ func TestSessionReconnectIsTrackedByClose(t *testing.T) { t.Fatal("session-triggered reconnect did not start") } - clt.Close() + closeDone := make(chan struct{}) + go func() { + clt.Close() + close(closeDone) + }() select { - case <-sessionStopDone: - case <-time.After(100 * time.Millisecond): - select { - case <-sessionStopDone: - case <-time.After(3 * time.Second): - t.Fatal("session stop did not return after reconnect backoff elapsed") - } + case <-closeDone: + close(releaseReconnect) t.Fatal("Close returned while the session-triggered reconnect was still running") + case <-time.After(100 * time.Millisecond): + } + + close(releaseReconnect) + select { + case <-closeDone: + case <-time.After(time.Second): + t.Fatal("Close did not return after the session-triggered reconnect completed") + } + select { + case <-sessionStopDone: + case <-time.After(time.Second): + t.Fatal("session stop did not return") } } diff --git a/transport/connection_test.go b/transport/connection_test.go index dfb9aa15..6a3f1aeb 100644 --- a/transport/connection_test.go +++ b/transport/connection_test.go @@ -86,13 +86,9 @@ func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) { entered: make(chan struct{}, 1), release: make(chan struct{}), } - defer func() { - select { - case <-underlying.release: - default: - close(underlying.release) - } - }() + var releaseOnce sync.Once + release := func() { releaseOnce.Do(func() { close(underlying.release) }) } + defer release() writer := newSnappyWriteFlusher(snappy.NewBufferedWriter(underlying)) writeDone := make(chan error, 1) @@ -118,7 +114,7 @@ func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) { case <-time.After(50 * time.Millisecond): } - close(underlying.release) + release() if err := <-writeDone; err != nil { t.Fatalf("Write failed: %v", err) } diff --git a/transport/server.go b/transport/server.go index 3008462c..f5f60808 100644 --- a/transport/server.go +++ b/transport/server.go @@ -48,6 +48,7 @@ import ( var ( errSelfConnect = perrors.New("connect self!") + errServerClosed = perrors.New("server closed") serverFastFailTimeout = time.Second * 1 serverID uatomic.Int32 @@ -198,6 +199,9 @@ func (s *server) listenTCP() error { err error streamListener net.Listener ) + if s.IsClosed() { + return errServerClosed + } if len(s.addr) == 0 || !strings.Contains(s.addr, ":") { streamListener, err = gxnet.ListenOnTCPRandomPort(s.addr) @@ -225,8 +229,19 @@ func (s *server) listenTCP() error { } } + return s.publishStreamListener(streamListener) +} + +func (s *server) publishStreamListener(streamListener net.Listener) error { addr := streamListener.Addr().String() s.lock.Lock() + select { + case <-s.done: + s.lock.Unlock() + _ = streamListener.Close() + return errServerClosed + default: + } s.streamListener = streamListener s.addr = addr s.lock.Unlock() @@ -240,6 +255,9 @@ func (s *server) listenUDP() error { localAddr *net.UDPAddr pktListener *net.UDPConn ) + if s.IsClosed() { + return errServerClosed + } if len(s.addr) == 0 || !strings.Contains(s.addr, ":") { pktListener, err = gxnet.ListenOnUDPRandomPort(s.addr) @@ -257,8 +275,19 @@ func (s *server) listenUDP() error { } } + return s.publishPacketListener(pktListener) +} + +func (s *server) publishPacketListener(pktListener net.PacketConn) error { addr := pktListener.LocalAddr().String() s.lock.Lock() + select { + case <-s.done: + s.lock.Unlock() + _ = pktListener.Close() + return errServerClosed + default: + } s.pktListener = pktListener s.addr = addr s.lock.Unlock() @@ -512,6 +541,9 @@ func (s *server) runWSSEventLoop(newSession NewSessionCallback) { // @newSession: new connection callback func (s *server) RunEventLoop(newSession NewSessionCallback) { if err := s.listen(); err != nil { + if perrors.Cause(err) == errServerClosed { + return + } panic(fmt.Errorf("server.listen() = error:%+v", perrors.WithStack(err))) } diff --git a/transport/server_test.go b/transport/server_test.go index c9a94ef9..605981a5 100644 --- a/transport/server_test.go +++ b/transport/server_test.go @@ -213,6 +213,91 @@ func TestServerCloseKeepsPublishedListener(t *testing.T) { }) } +func TestServerClosePreventsLateListenerPublication(t *testing.T) { + t.Run("already closed", func(t *testing.T) { + server := newServer(TCP_SERVER, WithLocalAddress("127.0.0.1:0")) + server.Close() + server.RunEventLoop(func(Session) error { return nil }) + if server.Listener() != nil { + t.Fatal("closed server opened a listener") + } + }) + + t.Run("TCP", func(t *testing.T) { + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := newServer(TCP_SERVER) + server.lock.Lock() + closeDone := make(chan struct{}) + go func() { + server.Close() + close(closeDone) + }() + select { + case <-server.done: + case <-time.After(time.Second): + server.lock.Unlock() + t.Fatal("Close did not start shutdown") + } + + publishDone := make(chan error, 1) + go func() { + publishDone <- server.publishStreamListener(listener) + }() + server.lock.Unlock() + + if err := <-publishDone; !errors.Is(err, errServerClosed) { + t.Fatalf("publishStreamListener returned %v, want %v", err, errServerClosed) + } + <-closeDone + if server.Listener() != nil { + t.Fatal("closed server published a late TCP listener") + } + if _, err := listener.Accept(); err == nil { + t.Fatal("late TCP listener remained open after rejected publication") + } + }) + + t.Run("UDP", func(t *testing.T) { + listener, err := net.ListenPacket("udp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := newServer(UDP_ENDPOINT) + server.lock.Lock() + closeDone := make(chan struct{}) + go func() { + server.Close() + close(closeDone) + }() + select { + case <-server.done: + case <-time.After(time.Second): + server.lock.Unlock() + t.Fatal("Close did not start shutdown") + } + + publishDone := make(chan error, 1) + go func() { + publishDone <- server.publishPacketListener(listener) + }() + server.lock.Unlock() + + if err := <-publishDone; !errors.Is(err, errServerClosed) { + t.Fatalf("publishPacketListener returned %v, want %v", err, errServerClosed) + } + <-closeDone + if server.PacketConn() != nil { + t.Fatal("closed server published a late UDP listener") + } + if _, _, err := listener.ReadFrom(make([]byte, 1)); err == nil { + t.Fatal("late UDP listener remained open after rejected publication") + } + }) +} + func TestServer(t *testing.T) { var addr string diff --git a/transport/session_test.go b/transport/session_test.go index 645c85a7..c7e5e3f8 100644 --- a/transport/session_test.go +++ b/transport/session_test.go @@ -22,7 +22,6 @@ import ( "io" "net" "sync" - "sync/atomic" "testing" "time" ) @@ -131,7 +130,6 @@ type resetBarrierNetConn struct { release chan struct{} enteredOnce sync.Once releaseOnce sync.Once - deadlines atomic.Int32 } func (c *resetBarrierNetConn) Read([]byte) (int, error) { @@ -140,19 +138,18 @@ func (c *resetBarrierNetConn) Read([]byte) (int, error) { return 0, io.EOF } -func (*resetBarrierNetConn) Write(p []byte) (int, error) { return len(p), nil } -func (*resetBarrierNetConn) Close() error { return nil } -func (*resetBarrierNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } -func (*resetBarrierNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } -func (*resetBarrierNetConn) SetDeadline(time.Time) error { return nil } -func (c *resetBarrierNetConn) SetReadDeadline(time.Time) error { - if c.deadlines.Add(1) > 1 { - c.releaseOnce.Do(func() { close(c.release) }) - } - return nil -} +func (*resetBarrierNetConn) Write(p []byte) (int, error) { return len(p), nil } +func (*resetBarrierNetConn) Close() error { return nil } +func (*resetBarrierNetConn) LocalAddr() net.Addr { return &net.TCPAddr{} } +func (*resetBarrierNetConn) RemoteAddr() net.Addr { return &net.TCPAddr{} } +func (*resetBarrierNetConn) SetDeadline(time.Time) error { return nil } +func (*resetBarrierNetConn) SetReadDeadline(time.Time) error { return nil } func (*resetBarrierNetConn) SetWriteDeadline(time.Time) error { return nil } +func (c *resetBarrierNetConn) releaseRead() { + c.releaseOnce.Do(func() { close(c.release) }) +} + func TestConcurrentWritePkgTimeoutRestoration(t *testing.T) { netConn := &timeoutTestNetConn{entered: make(chan *timeoutTestCall, 2)} ss := newTCPSession(netConn, nil).(*session) @@ -242,6 +239,7 @@ func TestResetWaitsForPackageLoop(t *testing.T) { } ss.Close() + netConn.releaseRead() select { case <-resetDone: case <-time.After(time.Second): diff --git a/transport/tls_test.go b/transport/tls_test.go index 485b34e0..630e55a7 100644 --- a/transport/tls_test.go +++ b/transport/tls_test.go @@ -50,4 +50,41 @@ func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { if config.MinVersion != tls.VersionTLS12 { t.Fatalf("MinVersion = %d, want TLS 1.2 (%d)", config.MinVersion, tls.VersionTLS12) } + if config.InsecureSkipVerify { + t.Fatal("InsecureSkipVerify is true; certificate verification must stay enabled") + } + if config.RootCAs == nil { + t.Fatal("RootCAs is nil; the configured trust collection was not loaded") + } + if len(config.Certificates) != 1 { + t.Fatalf("Certificates contains %d entries, want 1", len(config.Certificates)) + } +} + +func TestClientTLSConfigBuilderRejectsInvalidTrustCollection(t *testing.T) { + tempDir := t.TempDir() + certPath := filepath.Join(tempDir, "client.crt") + keyPath := filepath.Join(tempDir, "client.key") + caPath := filepath.Join(tempDir, "ca.crt") + for path, data := range map[string][]byte{ + certPath: WssServerCRT, + keyPath: WssServerKEY, + caPath: []byte("not a certificate"), + } { + if err := os.WriteFile(path, data, 0o600); err != nil { + t.Fatal(err) + } + } + + config, err := (&ClientTlsConfigBuilder{ + ClientKeyCertChainPath: certPath, + ClientPrivateKeyPath: keyPath, + ClientTrustCertCollectionPath: caPath, + }).BuildTlsConfig() + if err == nil { + t.Fatal("invalid trust collection returned nil error") + } + if config != nil { + t.Fatal("config must be nil when BuildTlsConfig fails") + } } From d864afc24ad42c446e408c8cc98829a82c6d8410 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 1 Aug 2026 07:10:29 +0800 Subject: [PATCH 6/8] Verify loaded TLS certificate contents Strengthen the client TLS builder regression test so it compares the configured trust pool and certificate chain against the requested PEM fixtures, preventing allocation-only assertions from masking wrong or empty certificate data. Constraint: Keep production TLS behavior unchanged and limit the change to review-requested test evidence. Confidence: High; the certificate-content assertions pass repeatedly with the project Go 1.25 toolchain. Scope-risk: Test-only change in transport/tls_test.go. Tested: WSL Go 1.25 targeted TLS builder tests with count=10; gofmt -d; git diff --check. Not-tested: GitHub CI and CodeRabbit on this new commit are pending. Co-authored-by: OmX --- transport/tls_test.go | 20 ++++++++++++++++++++ 1 file changed, 20 insertions(+) diff --git a/transport/tls_test.go b/transport/tls_test.go index 630e55a7..1c19dcdc 100644 --- a/transport/tls_test.go +++ b/transport/tls_test.go @@ -18,7 +18,10 @@ package getty import ( + "bytes" "crypto/tls" + "crypto/x509" + "encoding/pem" "os" "path/filepath" "testing" @@ -59,6 +62,23 @@ func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { if len(config.Certificates) != 1 { t.Fatalf("Certificates contains %d entries, want 1", len(config.Certificates)) } + expectedRootCAs := x509.NewCertPool() + if !expectedRootCAs.AppendCertsFromPEM(WssClientCRT) { + t.Fatal("failed to parse the expected root certificate") + } + if !config.RootCAs.Equal(expectedRootCAs) { + t.Fatal("RootCAs does not contain the configured trust certificate") + } + expectedClientCertificate, _ := pem.Decode(WssServerCRT) + if expectedClientCertificate == nil { + t.Fatal("failed to decode the expected client certificate") + } + if len(config.Certificates[0].Certificate) == 0 { + t.Fatal("configured client certificate has an empty certificate chain") + } + if !bytes.Equal(config.Certificates[0].Certificate[0], expectedClientCertificate.Bytes) { + t.Fatal("configured client certificate does not match the requested certificate") + } } func TestClientTLSConfigBuilderRejectsInvalidTrustCollection(t *testing.T) { From f654f82f4109d6d38ee576c6c945019b66598312 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 1 Aug 2026 07:19:22 +0800 Subject: [PATCH 7/8] Use a distinct TLS trust test fixture Replace the duplicated trust certificate in the TLS builder regression test with a separate self-signed root and assert that its DER differs from the client certificate. This makes the content checks fail if certificate and trust inputs are swapped. Constraint: Leave shared WSS integration fixtures and production TLS behavior unchanged. Confidence: High; repository fixtures were fingerprinted as identical and the revised assertions pass repeatedly with distinct certificate data. Scope-risk: Test-only fixture and assertions in transport/tls_test.go. Tested: WSL Go 1.25 targeted TLS builder tests with count=10; gofmt -d; git diff --check. Not-tested: GitHub CI and CodeRabbit on this new commit are pending. Co-authored-by: OmX --- transport/tls_test.go | 23 +++++++++++++++++++++-- 1 file changed, 21 insertions(+), 2 deletions(-) diff --git a/transport/tls_test.go b/transport/tls_test.go index 1c19dcdc..29fb4d68 100644 --- a/transport/tls_test.go +++ b/transport/tls_test.go @@ -27,6 +27,18 @@ import ( "testing" ) +var tlsTestRootCertificate = []byte(`-----BEGIN CERTIFICATE----- +MIIBiDCCAS+gAwIBAgIUMaJuA5AGTTBvqSWb4fhJCC7UY4wwCgYIKoZIzj0EAwIw +GjEYMBYGA1UEAwwPZ2V0dHktdGVzdC1yb290MB4XDTI2MDczMTIzMTYyN1oXDTM2 +MDcyODIzMTYyN1owGjEYMBYGA1UEAwwPZ2V0dHktdGVzdC1yb290MFkwEwYHKoZI +zj0CAQYIKoZIzj0DAQcDQgAEWZNS+42M+wb2AmNunl7ccsdoaRYanWn1kgt5Rj7X +50hqE1aA8Wdl7dbbDmCwSrwLRNus1ebi2571N0XJNXn536NTMFEwHQYDVR0OBBYE +FJYJbIsdqMVkz65eVtuLmz41l4IjMB8GA1UdIwQYMBaAFJYJbIsdqMVkz65eVtuL +mz41l4IjMA8GA1UdEwEB/wQFMAMBAf8wCgYIKoZIzj0EAwIDRwAwRAIgX6EFP2GN +UF0MEbozG6tzqvrF1R8NUNEUEF4ThXnQMpMCIB/191gSjtjhiuDKu/pT5cCXe9ka +Wf17jc2sFoJ9DUsb +-----END CERTIFICATE-----`) + func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { tempDir := t.TempDir() certPath := filepath.Join(tempDir, "client.crt") @@ -35,7 +47,7 @@ func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { for path, data := range map[string][]byte{ certPath: WssServerCRT, keyPath: WssServerKEY, - caPath: WssClientCRT, + caPath: tlsTestRootCertificate, } { if err := os.WriteFile(path, data, 0o600); err != nil { t.Fatal(err) @@ -63,7 +75,7 @@ func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { t.Fatalf("Certificates contains %d entries, want 1", len(config.Certificates)) } expectedRootCAs := x509.NewCertPool() - if !expectedRootCAs.AppendCertsFromPEM(WssClientCRT) { + if !expectedRootCAs.AppendCertsFromPEM(tlsTestRootCertificate) { t.Fatal("failed to parse the expected root certificate") } if !config.RootCAs.Equal(expectedRootCAs) { @@ -73,6 +85,13 @@ func TestClientTLSConfigBuilderMinimumVersion(t *testing.T) { if expectedClientCertificate == nil { t.Fatal("failed to decode the expected client certificate") } + expectedRootCertificate, _ := pem.Decode(tlsTestRootCertificate) + if expectedRootCertificate == nil { + t.Fatal("failed to decode the expected root certificate") + } + if bytes.Equal(expectedClientCertificate.Bytes, expectedRootCertificate.Bytes) { + t.Fatal("client and root certificate fixtures must be distinct") + } if len(config.Certificates[0].Certificate) == 0 { t.Fatal("configured client certificate has an empty certificate chain") } From 341c27600af3195cb707125f4969d78bf3939e17 Mon Sep 17 00:00:00 2001 From: "Xin.Zh" Date: Sat, 1 Aug 2026 07:34:28 +0800 Subject: [PATCH 8/8] Report consumed bytes when flush fails Return the byte count accepted by the flate and snappy writers when their subsequent flush fails. This preserves the delivery error while honoring io.Writer byte-count semantics and avoids inviting callers to resend already-consumed input. Constraint: Keep compression, flushing, locking, and error wrapping behavior unchanged. Confidence: High; both implementations failed the new regression test before the change and pass after the two-line production fix. Scope-risk: Limited to error-path byte counts for compressed connection writes. Tested: Windows targeted test count=10; WSL Go 1.25 targeted test -race count=10; go test ./...; go vet ./transport ./util; go build ./transport ./util; gofmt -d; git diff --check. Not-tested: GitHub CI and CodeRabbit on this new commit are pending. Co-authored-by: OmX --- transport/connection.go | 4 ++-- transport/connection_test.go | 45 ++++++++++++++++++++++++++++++++++++ 2 files changed, 47 insertions(+), 2 deletions(-) diff --git a/transport/connection.go b/transport/connection.go index 091fe18b..4fd89023 100644 --- a/transport/connection.go +++ b/transport/connection.go @@ -227,7 +227,7 @@ func (t *writeFlusher) Write(p []byte) (int, error) { return n, perrors.WithStack(err) } if err := t.flusher.Flush(); err != nil { - return 0, perrors.WithStack(err) + return n, perrors.WithStack(err) } return n, nil @@ -254,7 +254,7 @@ func (s *snappyWriteFlusher) Write(p []byte) (int, error) { return n, perrors.WithStack(err) } if err := s.writer.Flush(); err != nil { - return 0, perrors.WithStack(err) + return n, perrors.WithStack(err) } return n, nil } diff --git a/transport/connection_test.go b/transport/connection_test.go index 6a3f1aeb..ca601f3e 100644 --- a/transport/connection_test.go +++ b/transport/connection_test.go @@ -18,6 +18,8 @@ package getty import ( + "compress/flate" + "errors" "io" "net" "sync" @@ -34,6 +36,14 @@ type blockingSnappyWriter struct { release chan struct{} } +var errFlushWriter = errors.New("flush writer failure") + +type flushErrorWriter struct{} + +func (flushErrorWriter) Write([]byte) (int, error) { + return 0, errFlushWriter +} + func (w *blockingSnappyWriter) Write(p []byte) (int, error) { select { case w.entered <- struct{}{}: @@ -81,6 +91,41 @@ func TestConnectionTimeoutAccessorsDoNotCopyAtomicState(t *testing.T) { wg.Wait() } +func TestWriteFlushersReturnConsumedBytesOnFlushError(t *testing.T) { + payload := []byte("payload") + tests := []struct { + name string + writer io.Writer + }{ + { + name: "flate", + writer: func() io.Writer { + writer, err := flate.NewWriter(flushErrorWriter{}, flate.DefaultCompression) + if err != nil { + t.Fatal(err) + } + return &writeFlusher{flusher: writer} + }(), + }, + { + name: "snappy", + writer: newSnappyWriteFlusher(snappy.NewBufferedWriter(flushErrorWriter{})), + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + n, err := test.writer.Write(payload) + if !errors.Is(err, errFlushWriter) { + t.Fatalf("Write error = %v, want %v", err, errFlushWriter) + } + if n != len(payload) { + t.Fatalf("Write returned %d bytes after consuming %d", n, len(payload)) + } + }) + } +} + func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) { underlying := &blockingSnappyWriter{ entered: make(chan struct{}, 1),