From e060c93ea3f9f808f32acc74dc51abca75b4b0ea Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Mon, 10 Aug 2026 21:02:01 +0800
Subject: [PATCH 01/15] =?UTF-8?q?fix/setcompression=E5=90=8E=E5=87=BA?=
 =?UTF-8?q?=E7=8E=B0=E5=9D=8F=E6=B5=81?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 transport/connection.go | 11 ++++-------
 1 file changed, 4 insertions(+), 7 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 4fd89023..394bad65 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -25,19 +25,14 @@ import (
 	"net"
 	"sync"
 	"time"
-)
 
-import (
 	"github.com/golang/snappy"
-
 	"github.com/gorilla/websocket"
 
 	perrors "github.com/pkg/errors"
 
 	uatomic "go.uber.org/atomic"
-)
 
-import (
 	log "github.com/AlexStocks/getty/util"
 )
 
@@ -345,8 +340,10 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		// 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)
+			if _, isRaw := t.writer.(net.Conn); isRaw {
+				netBuf := net.Buffers(buffers)
+				lg, err = netBuf.WriteTo(t.conn)
+			}
 		} else {
 			for _, b := range buffers {
 				var n int

From 3b4ed182468b4ce6a7202d0943691ea967fdc575 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Mon, 10 Aug 2026 21:53:43 +0800
Subject: [PATCH 02/15] feat/Add batch flush interface

---
 transport/connection.go | 74 ++++++++++++++++++++++++++++++++++++-----
 1 file changed, 65 insertions(+), 9 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 394bad65..2aac0a0f 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -76,8 +76,14 @@ type Connection interface {
 // ///////////////////////////////////////
 
 type gettyConn struct {
-	id            uint32
-	compress      CompressType
+	id       uint32
+	compress CompressType
+	// isCompressed reports whether reader/writer have been replaced by a codec.
+	// compress == CompressNone is not a substitute: CompressNone is
+	// flate.NoCompression(0), so SetCompressType(CompressNone) still installs a
+	// flate codec that frames the stream into deflate blocks. Such a stream is
+	// stateful and must not be treated like a raw connection.
+	isCompressed  bool
 	readBytes     uatomic.Uint32   // read bytes
 	writeBytes    uatomic.Uint32   // write bytes
 	readPkgNum    uatomic.Uint32   // send pkg number
@@ -204,6 +210,13 @@ func newGettyTCPConn(conn net.Conn) *gettyTCPConn {
 	}
 }
 
+// buffersWriter is implemented by writers that can take a whole batch at once.
+// The compress writers implement it so a batch becomes one compressed block
+// instead of one block per packet.
+type buffersWriter interface {
+	WriteBuffers(buffers [][]byte) (int64, error)
+}
+
 // for zip compress
 type writeFlusher struct {
 	flusher *flate.Writer
@@ -228,6 +241,25 @@ func (t *writeFlusher) Write(p []byte) (int, error) {
 	return n, nil
 }
 
+// WriteBuffers writes the whole batch into the compressor and flushes once at
+// the end. Calling Write per buffer would flush every time and degrade into N
+// separate deflate blocks: N syscalls, and no way to exploit patterns that
+// repeat across the packets.
+func (t *writeFlusher) WriteBuffers(buffers [][]byte) (int64, error) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+
+	var total int64
+	for _, b := range buffers {
+		n, err := t.flusher.Write(b)
+		total += int64(n)
+		if err != nil {
+			return total, perrors.WithStack(err)
+		}
+	}
+	return total, perrors.WithStack(t.flusher.Flush())
+}
+
 // 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
@@ -254,6 +286,23 @@ func (s *snappyWriteFlusher) Write(p []byte) (int, error) {
 	return n, nil
 }
 
+// WriteBuffers mirrors writeFlusher.WriteBuffers: write the whole batch, flush
+// once at the end, so a batch leaves as a single snappy block.
+func (s *snappyWriteFlusher) WriteBuffers(buffers [][]byte) (int64, error) {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	var total int64
+	for _, b := range buffers {
+		n, err := s.writer.Write(b)
+		total += int64(n)
+		if err != nil {
+			return total, perrors.WithStack(err)
+		}
+	}
+	return total, perrors.WithStack(s.writer.Flush())
+}
+
 func (s *snappyWriteFlusher) Close() error {
 	s.lock.Lock()
 	defer s.lock.Unlock()
@@ -285,6 +334,10 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
+	// Both branches replaced reader/writer with a codec (CompressNone included,
+	// see the isCompressed comment), so the conn is no longer raw: no deadlines,
+	// and all IO must go through t.reader/t.writer.
+	t.isCompressed = true
 	t.compress = c
 }
 
@@ -297,7 +350,9 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	)
 
 	// set read timeout deadline
-	if t.compress == CompressNone && t.rTimeout.Load() > 0 {
+	// No deadline on a codec stream: one timeout leaves half a block in the
+	// decoder and every byte after it is misaligned.
+	if !t.isCompressed && t.rTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -324,7 +379,7 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		lg          int64
 	)
 
-	if t.compress == CompressNone && t.wTimeout.Load() > 0 {
+	if !t.isCompressed && t.wTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -339,11 +394,12 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		// 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 {
-			if _, isRaw := t.writer.(net.Conn); isRaw {
-				netBuf := net.Buffers(buffers)
-				lg, err = netBuf.WriteTo(t.conn)
-			}
+		if !t.isCompressed {
+			// only a raw conn here, so writev the whole batch in one syscall.
+			netBuf := net.Buffers(buffers)
+			lg, err = netBuf.WriteTo(t.conn)
+		} else if bw, ok := t.writer.(buffersWriter); ok {
+			lg, err = bw.WriteBuffers(buffers)
 		} else {
 			for _, b := range buffers {
 				var n int

From 8900f19d470a68f82c9a1d1bb945cd95286b17b9 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Tue, 11 Aug 2026 21:01:30 +0800
Subject: [PATCH 03/15] fix: route [][]byte sends through codec writer and
 clarify codec state

- rename isCompressed to codecEnabled; SetCompressType(CompressNone)
  still installs a flate codec stream, so the state must not be named
  by compression level
- keep batch [][]byte sends on the codec writer (single flush) so a
  codec connection never mixes coded and raw frames (#102/#107)
- add wire-format regression tests: mixed []byte and [][]byte sends
  over a CompressNone codec pair, plus raw writev coverage when
  SetCompressType was never called
- make TestTCPClient's dummy peer a TCP discard server so codec frames
  do not make an HTTP peer close mid-test
- run make fmt; imports are formatter-clean
---
 transport/client_test.go     |  19 ++++-
 transport/connection.go      |  26 ++++---
 transport/connection_test.go | 140 +++++++++++++++++++++++++++++++++++
 3 files changed, 172 insertions(+), 13 deletions(-)

diff --git a/transport/client_test.go b/transport/client_test.go
index 5b8afd95..cb38f096 100644
--- a/transport/client_test.go
+++ b/transport/client_test.go
@@ -21,8 +21,8 @@ import (
 	"bytes"
 	"crypto/tls"
 	"errors"
+	"io"
 	"net"
-	"net/http"
 	"os"
 	"path/filepath"
 	"strconv"
@@ -250,7 +250,22 @@ func TestTCPClient(t *testing.T) {
 			return nil, err
 		}
 
-		go func() { _ = http.Serve(listener, nil) }()
+		// Read and discard everything the client sends. The peer must not
+		// interpret the bytes as HTTP: once the codec batch path is enabled,
+		// it receives deflate/snappy frames, and an HTTP server closes the
+		// connection mid-test, making the write accounting assertions flaky.
+		go func() {
+			for {
+				conn, err := listener.Accept()
+				if err != nil {
+					return
+				}
+				go func() {
+					defer func() { _ = conn.Close() }()
+					_, _ = io.Copy(io.Discard, conn)
+				}()
+			}
+		}()
 		return listener, nil
 	}
 
diff --git a/transport/connection.go b/transport/connection.go
index 2aac0a0f..02cc910b 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -25,14 +25,19 @@ import (
 	"net"
 	"sync"
 	"time"
+)
 
+import (
 	"github.com/golang/snappy"
+
 	"github.com/gorilla/websocket"
 
 	perrors "github.com/pkg/errors"
 
 	uatomic "go.uber.org/atomic"
+)
 
+import (
 	log "github.com/AlexStocks/getty/util"
 )
 
@@ -78,12 +83,12 @@ type Connection interface {
 type gettyConn struct {
 	id       uint32
 	compress CompressType
-	// isCompressed reports whether reader/writer have been replaced by a codec.
+	// codecEnabled reports whether reader/writer have been replaced by a codec.
 	// compress == CompressNone is not a substitute: CompressNone is
 	// flate.NoCompression(0), so SetCompressType(CompressNone) still installs a
 	// flate codec that frames the stream into deflate blocks. Such a stream is
 	// stateful and must not be treated like a raw connection.
-	isCompressed  bool
+	codecEnabled  bool
 	readBytes     uatomic.Uint32   // read bytes
 	writeBytes    uatomic.Uint32   // write bytes
 	readPkgNum    uatomic.Uint32   // send pkg number
@@ -335,9 +340,9 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
 	// Both branches replaced reader/writer with a codec (CompressNone included,
-	// see the isCompressed comment), so the conn is no longer raw: no deadlines,
+	// see the codecEnabled comment), so the conn is no longer raw: no deadlines,
 	// and all IO must go through t.reader/t.writer.
-	t.isCompressed = true
+	t.codecEnabled = true
 	t.compress = c
 }
 
@@ -352,7 +357,7 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	// set read timeout deadline
 	// No deadline on a codec stream: one timeout leaves half a block in the
 	// decoder and every byte after it is misaligned.
-	if !t.isCompressed && t.rTimeout.Load() > 0 {
+	if !t.codecEnabled && t.rTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -379,7 +384,7 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		lg          int64
 	)
 
-	if !t.isCompressed && t.wTimeout.Load() > 0 {
+	if !t.codecEnabled && t.wTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -390,11 +395,10 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 	}
 
 	if buffers, ok := pkg.([][]byte); ok {
-		// #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.isCompressed {
+		// #102: when a codec is installed the [][]byte path must go through
+		// t.writer (the codec writer), otherwise it writes raw frames directly
+		// to t.conn and the peer receives a corrupt mix of coded and raw data.
+		if !t.codecEnabled {
 			// only a raw conn here, so writev the whole batch in one syscall.
 			netBuf := net.Buffers(buffers)
 			lg, err = netBuf.WriteTo(t.conn)
diff --git a/transport/connection_test.go b/transport/connection_test.go
index ca601f3e..57736aa1 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -18,6 +18,7 @@
 package getty
 
 import (
+	"bytes"
 	"compress/flate"
 	"errors"
 	"io"
@@ -167,3 +168,142 @@ func TestSnappyWriteFlusherCloseWaitsForWrite(t *testing.T) {
 		t.Fatalf("Close failed: %v", err)
 	}
 }
+
+func newTCPConnPair(t *testing.T) (*gettyTCPConn, *gettyTCPConn) {
+	t.Helper()
+
+	listener, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer func() { _ = listener.Close() }()
+
+	type acceptResult struct {
+		conn net.Conn
+		err  error
+	}
+	accepted := make(chan acceptResult, 1)
+	go func() {
+		conn, err := listener.Accept()
+		accepted <- acceptResult{conn: conn, err: err}
+	}()
+
+	clientRaw, err := net.Dial("tcp", listener.Addr().String())
+	if err != nil {
+		t.Fatal(err)
+	}
+	result := <-accepted
+	if result.err != nil {
+		_ = clientRaw.Close()
+		t.Fatal(result.err)
+	}
+
+	client := newGettyTCPConn(clientRaw)
+	server := newGettyTCPConn(result.conn)
+	t.Cleanup(func() {
+		client.CloseConn(0)
+		server.CloseConn(0)
+	})
+	return client, server
+}
+
+// startReceiver continuously decodes from conn until wantLen bytes have been
+// read. A tiny buffer forces reads to cross codec block boundaries.
+func startReceiver(conn *gettyTCPConn, wantLen int, bufSize int) (<-chan []byte, <-chan error) {
+	gotCh := make(chan []byte, 1)
+	errCh := make(chan error, 1)
+	go func() {
+		var got []byte
+		buf := make([]byte, bufSize)
+		for len(got) < wantLen {
+			n, err := conn.recv(buf)
+			if err != nil {
+				errCh <- err
+				return
+			}
+			if n == 0 {
+				continue
+			}
+			got = append(got, buf[:n]...)
+		}
+		gotCh <- got
+	}()
+	return gotCh, errCh
+}
+
+// TestSendMixedSingleAndBatchOverCompressNoneCodec pins the #102/#107 wire
+// format: SetCompressType(CompressNone) installs a real flate codec, so every
+// Send path, including [][]byte, must go through the codec writer. Otherwise
+// one connection carries a corrupt mix of coded []byte sends and raw
+// [][]byte sends and the peer's decoder fails.
+func TestSendMixedSingleAndBatchOverCompressNoneCodec(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	client.SetCompressType(CompressNone)
+	server.SetCompressType(CompressNone)
+
+	single := []byte("single-packet-")
+	batch := [][]byte{
+		[]byte("batch-part-0-"),
+		[]byte("batch-part-1-"),
+		[]byte("batch-part-2"),
+	}
+	tail := []byte("-tail")
+	expected := bytes.Join([][]byte{
+		single,
+		batch[0],
+		batch[1],
+		batch[2],
+		tail,
+	}, nil)
+
+	gotCh, errCh := startReceiver(server, len(expected), 3)
+	for _, pkg := range []any{single, batch, tail} {
+		if _, err := client.Send(pkg); err != nil {
+			t.Fatalf("Send(%T) failed: %v", pkg, err)
+		}
+	}
+
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("decoded stream mismatch:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to decode codec stream: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the decoded stream")
+	}
+}
+
+// TestSendBatchRawWithoutCompressType keeps the raw writev path covered: when
+// SetCompressType was never called, a [][]byte send must reach the peer
+// untouched (net.Buffers.WriteTo(t.conn), a single writev on *net.TCPConn).
+func TestSendBatchRawWithoutCompressType(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	if client.codecEnabled {
+		t.Fatal("new connection unexpectedly has a codec installed")
+	}
+
+	batch := [][]byte{
+		[]byte("raw-part-0-"),
+		[]byte("raw-part-1-"),
+		[]byte("raw-part-2"),
+	}
+	expected := bytes.Join(batch, nil)
+
+	gotCh, errCh := startReceiver(server, len(expected), 3)
+	if _, err := client.Send(batch); err != nil {
+		t.Fatalf("Send([][]byte) failed: %v", err)
+	}
+
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("raw stream mismatch:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to read raw stream: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the raw stream")
+	}
+}

From 604c153af9405b7888c35f1ff2b11d09a8c395c3 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Wed, 12 Aug 2026 22:27:10 +0800
Subject: [PATCH 04/15] fix: keep deadlines on codec streams and terminate
 stalled connections

- arm a read deadline on codec streams via CodecStallTimeout (wider than
  the raw rTimeout poll interval) so a peer that sends half a codec block
  cannot block recv forever; the write deadline now always reaches the
  socket, so SetWriteTimeout and WritePkg(pkg, timeout) work on codec
  connections
- on a codec timeout, latch the stream as broken and close the socket:
  recv/Send fail fast with ErrCodecStreamBroken, the session treats it as
  fatal and closes/reconnects instead of retrying a dead decoder
- keep normal shutdown working: session.stop()'s unblock deadline is
  passed through on the read path, and CloseConn skips flushing a broken
  snappy writer
- add stalled peer read/write regression tests (flate and snappy) that
  verify the call returns within the deadline, the codec connection is
  actually terminated, and idle/clean-close behavior is preserved
---
 transport/connection.go      | 167 ++++++++++++++++++--
 transport/connection_test.go | 285 +++++++++++++++++++++++++++++++++++
 2 files changed, 439 insertions(+), 13 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 02cc910b..a3dcc035 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -20,9 +20,11 @@ package getty
 import (
 	"compress/flate"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"net"
+	"os"
 	"sync"
 	"time"
 )
@@ -46,6 +48,33 @@ var (
 	connID     uatomic.Uint32
 )
 
+// ErrCodecStreamBroken is returned once a codec(compressed) stream cannot be
+// trusted any more: a read/write timed out or failed in the middle of a codec
+// block, so the decoder/encoder state and the bytes already on the wire are
+// desynchronized. flate/snappy latch such errors forever, hence the connection
+// is unusable and the session must be closed and rebuilt. It deliberately does
+// not unwrap to a net.Error, so session.handleTCPPackage treats it as fatal
+// instead of as a benign read timeout that can be retried.
+var ErrCodecStreamBroken = errors.New("getty: codec stream is broken, connection must be closed")
+
+// CodecStallTimeout is the read deadline used on a codec(compressed) stream,
+// and thus the upper bound on how long a stalled peer can block a read.
+//
+// It exists because the connection read timeout (see SetReadTimeout, 1s by
+// default) is a poll interval, not an idle timeout: session.handleTCPPackage
+// retries after every read timeout. A codec stream cannot be retried
+// (flate/snappy latch read errors forever), so arming rTimeout on it would tear
+// down every compressed connection that stays idle for one second. Hence codec
+// reads get this much wider deadline instead, and hitting it means the peer
+// really stalled mid-stream: the connection is declared broken.
+//
+// Keep it well above the read timeout and above the application heartbeat
+// period, otherwise idle compressed connections get killed. Set it to 0 to arm
+// no read deadline at all on codec streams (a stalled peer then blocks until
+// the session is closed). It is read once, when SetCompressType installs the
+// codec, so set it before creating connections.
+var CodecStallTimeout = 5 * time.Minute
+
 // Connection wrap some connection params and operations
 type Connection interface {
 	ID() uint32
@@ -184,6 +213,14 @@ type gettyTCPConn struct {
 	reader io.Reader
 	writer io.Writer
 	conn   net.Conn
+	// codecBroken is set once a codec read/write failed in the middle of the
+	// stream. Written by the read goroutine and read by writer goroutines, so
+	// it has to be atomic. Once set, recv/Send fail fast: touching the codec
+	// again would only produce more garbage on the wire.
+	codecBroken uatomic.Bool
+	// codecStallTimeout is the per-conn copy of CodecStallTimeout, taken when
+	// the codec is installed. 0 arms no read deadline on the codec stream.
+	codecStallTimeout time.Duration
 }
 
 // create gettyTCPConn
@@ -340,12 +377,100 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
 	// Both branches replaced reader/writer with a codec (CompressNone included,
-	// see the codecEnabled comment), so the conn is no longer raw: no deadlines,
-	// and all IO must go through t.reader/t.writer.
+	// see the codecEnabled comment), so the conn is no longer raw: all IO must
+	// go through t.reader/t.writer, and a read/write error can no longer be
+	// retried on this stream.
 	t.codecEnabled = true
+	t.codecStallTimeout = CodecStallTimeout
 	t.compress = c
 }
 
+// readDeadlineTimeout returns the deadline to arm before a read.
+//
+// A raw conn uses rTimeout, whose timeouts session.handleTCPPackage simply
+// retries. A codec stream cannot be retried, so it gets CodecStallTimeout
+// instead - see that variable for why rTimeout is unusable here. The larger of
+// the two wins: raising rTimeout widens the codec bound, and lowering
+// CodecStallTimeout tightens it.
+func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
+	timeout := t.rTimeout.Load()
+	if !t.codecEnabled {
+		return timeout
+	}
+	if t.codecStallTimeout <= 0 {
+		// stall cap disabled: no deadline, the read blocks until data arrives
+		// or session.stop() arms a deadline to unblock it.
+		return 0
+	}
+	if timeout < t.codecStallTimeout {
+		return t.codecStallTimeout
+	}
+	return timeout
+}
+
+// codecIOError maps a timeout on a codec stream to the fatal
+// ErrCodecStreamBroken and latches the connection as broken.
+//
+// A timed out read leaves half a block in the decoder and misaligns every byte
+// after it; a timed out write leaves half a block on the wire and misaligns the
+// peer's decoder. Either way the stream cannot be reused, so the timeout is
+// reported as ErrCodecStreamBroken instead of as a net.Error: that is what makes
+// session.handleTCPPackage treat it as fatal and close the session (a client
+// then reconnects) rather than retry the read.
+//
+// The socket is closed as well, so the codec connection is terminated rather
+// than left half-alive: it unblocks any concurrent Read/Write and lets the
+// session tear down even when the stalled side is the writer. CloseConn is not
+// used here because closing the snappy writer can block on the very peer that
+// stalled; closing the raw conn is enough. t.conn is left in place (session.gc
+// nils it via CloseConn) so a concurrent recv/Send never sees a nil conn, and
+// CloseConn skips the broken writer when it eventually runs.
+//
+// Any other failure keeps its own identity - io.EOF above all, which the session
+// read loop matches on to detect a clean peer shutdown. Those need no latch:
+// flate/snappy record the failure internally and emit nothing more, every
+// further Read/Write just returns it again.
+func (t *gettyTCPConn) codecIOError(err error) error {
+	if err == nil || !t.codecEnabled || !isTimeoutError(err) {
+		return perrors.WithStack(err)
+	}
+
+	t.codecBroken.Store(true)
+	if conn := t.conn; conn != nil {
+		_ = conn.Close()
+	}
+	return perrors.Wrapf(ErrCodecStreamBroken, "codec stream stalled: %v", err)
+}
+
+// codecReadError is codecIOError for the read path, where a timeout raised while
+// the session is closing must be passed through: session.stop() deliberately
+// arms a read deadline to unblock this goroutine, and that timeout is a shutdown
+// signal, not a stalled peer. Latching it would make every normal close of a
+// compressed session report an error to the listener. The write path has no such
+// exemption - a timed out write desynchronizes the peer whether we are closing
+// or not.
+func (t *gettyTCPConn) codecReadError(err error) error {
+	if t.codecEnabled && isTimeoutError(err) && t.sessionClosing() {
+		return perrors.WithStack(err)
+	}
+	return t.codecIOError(err)
+}
+
+func (t *gettyTCPConn) sessionClosing() bool {
+	// t.ss is set by newSession before any IO goroutine starts; it is nil only
+	// for a connection used without a session (unit tests).
+	ss := t.ss
+	return ss != nil && ss.IsClosed()
+}
+
+func isTimeoutError(err error) bool {
+	var netErr net.Error
+	if errors.As(err, &netErr) && netErr.Timeout() {
+		return true
+	}
+	return errors.Is(err, os.ErrDeadlineExceeded)
+}
+
 // tcp connection read
 func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	var (
@@ -354,14 +479,16 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 		length      int
 	)
 
+	if t.codecBroken.Load() {
+		return 0, perrors.WithStack(ErrCodecStreamBroken)
+	}
+
 	// set read timeout deadline
-	// No deadline on a codec stream: one timeout leaves half a block in the
-	// decoder and every byte after it is misaligned.
-	if !t.codecEnabled && t.rTimeout.Load() > 0 {
+	if timeout := t.readDeadlineTimeout(); timeout > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
-		if err = t.conn.SetReadDeadline(currentTime.Add(t.rTimeout.Load())); err != nil {
+		if err = t.conn.SetReadDeadline(currentTime.Add(timeout)); err != nil {
 			// just a timeout error
 			return 0, perrors.WithStack(err)
 		}
@@ -370,7 +497,7 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 
 	length, err = t.reader.Read(p)
 	t.readBytes.Add(uint32(length))
-	return length, perrors.WithStack(err)
+	return length, t.codecReadError(err)
 }
 
 // tcp connection write
@@ -384,7 +511,16 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		lg          int64
 	)
 
-	if !t.codecEnabled && t.wTimeout.Load() > 0 {
+	if t.codecBroken.Load() {
+		return 0, perrors.WithStack(ErrCodecStreamBroken)
+	}
+
+	// The write deadline applies to a codec stream too: unlike a read, a write
+	// only blocks when the peer stopped reading, so it never fires on an idle
+	// connection and there is nothing to poll for. Skipping it here used to make
+	// SetWriteTimeout - and the per-call WritePkg(pkg, timeout) - silently
+	// ineffective, letting a stalled peer block writers forever.
+	if t.wTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -420,7 +556,7 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		}
 		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%s",
 			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
-		return int(lg), perrors.WithStack(err)
+		return int(lg), t.codecIOError(err)
 	}
 
 	if p, ok = pkg.([]byte); ok {
@@ -431,7 +567,7 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		}
 		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%v",
 			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
-		return length, perrors.WithStack(err)
+		return length, t.codecIOError(err)
 	}
 
 	return 0, perrors.Errorf("illegal @pkg{%#v} type", pkg)
@@ -445,9 +581,14 @@ 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.Close(); err != nil {
-				log.Errorf("snappy.Writer.Close() = error:%+v", err)
+		// A broken codec must not be flushed: the stream is already
+		// desynchronized, and Close would only push more garbage into a socket
+		// that may still be stalled.
+		if !t.codecBroken.Load() {
+			if writer, ok := t.writer.(*snappyWriteFlusher); ok {
+				if err := 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
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 57736aa1..8b0ba67a 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -30,6 +30,8 @@ import (
 
 import (
 	"github.com/golang/snappy"
+
+	perrors "github.com/pkg/errors"
 )
 
 type blockingSnappyWriter struct {
@@ -307,3 +309,286 @@ func TestSendBatchRawWithoutCompressType(t *testing.T) {
 		t.Fatal("timed out waiting for the raw stream")
 	}
 }
+
+// codecStallCases covers both codec families: they take different branches in
+// SetCompressType and both latch IO errors, so the stall handling has to hold
+// for each.
+var codecStallCases = []struct {
+	name     string
+	compress CompressType
+}{
+	{name: "flate", compress: CompressZip},
+	{name: "snappy", compress: CompressSnappy},
+}
+
+// halfCodecBlock returns the first half of a valid, flushed codec block: a prefix
+// the decoder cannot finish decoding, i.e. exactly what a peer that dies
+// mid-write leaves behind.
+func halfCodecBlock(t *testing.T, c CompressType, payload string) []byte {
+	t.Helper()
+
+	var (
+		buf    bytes.Buffer
+		writer interface {
+			io.Writer
+			Flush() error
+		}
+	)
+	if c == CompressSnappy {
+		writer = snappy.NewBufferedWriter(&buf)
+	} else {
+		flateWriter, err := flate.NewWriter(&buf, int(c))
+		if err != nil {
+			t.Fatal(err)
+		}
+		writer = flateWriter
+	}
+	if _, err := writer.Write([]byte(payload)); err != nil {
+		t.Fatal(err)
+	}
+	if err := writer.Flush(); err != nil {
+		t.Fatal(err)
+	}
+	block := buf.Bytes()
+	if len(block) < 4 {
+		t.Fatalf("codec block too small to truncate: %d bytes", len(block))
+	}
+	return block[:len(block)/2]
+}
+
+// assertCodecStreamBroken checks the error a stalled codec stream must produce.
+// The last two assertions are the actual encoding of the fix: session.handleTCPPackage
+// classifies read errors by perrors.Cause(), retrying net.Error timeouts and
+// treating io.EOF as a clean peer shutdown. A stalled codec stream must fall
+// into neither bucket, otherwise the session keeps reading from a decoder that
+// has already latched the error.
+func assertCodecStreamBroken(t *testing.T, err error) {
+	t.Helper()
+
+	if err == nil {
+		t.Fatal("stalled codec stream returned no error")
+	}
+	if !errors.Is(err, ErrCodecStreamBroken) {
+		t.Fatalf("error = %v, want ErrCodecStreamBroken", err)
+	}
+	cause := perrors.Cause(err)
+	if netErr, ok := cause.(net.Error); ok && netErr.Timeout() {
+		t.Fatalf("cause %v is a retryable net.Error timeout; the session would keep using the dead codec", cause)
+	}
+	if cause == io.EOF {
+		t.Fatalf("cause %v would be treated as a clean peer shutdown", cause)
+	}
+}
+
+// TestCodecRecvStalledPeerBreaksStream covers the P1: a peer that sends half a
+// codec block and then goes silent must not block the reader forever. The read
+// deadline has to reach the socket even though a codec is installed, and the
+// resulting timeout must terminate the connection instead of being retried on a
+// decoder that has already latched the error.
+func TestCodecRecvStalledPeerBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			if _, err := server.conn.Write(halfCodecBlock(t, test.compress, "stalled-peer-payload")); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+			// the peer now stalls: no more bytes, and the connection stays open.
+
+			// recv runs in a goroutine so a regression (no deadline on the codec
+			// stream) fails the test in seconds instead of hanging until the test
+			// binary panics.
+			recvErr := make(chan error, 1)
+			go func() {
+				buf := make([]byte, 64)
+				for callsLeft := 1000; callsLeft > 0; callsLeft-- {
+					if _, err := client.recv(buf); err != nil {
+						recvErr <- err
+						return
+					}
+				}
+				recvErr <- nil
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("recv never returned: the read deadline did not reach the socket")
+			}
+
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+			// the codec connection must be terminated, not just marked unusable:
+			// the peer has to observe the close instead of a silent open socket.
+			if err := client.conn.SetReadDeadline(time.Now()); err == nil {
+				t.Fatal("underlying conn is still open after a stalled read")
+			}
+			_ = server.conn.SetReadDeadline(time.Now().Add(time.Second))
+			if _, err := server.conn.Read(make([]byte, 8)); err == nil {
+				t.Fatal("peer read still succeeded after the codec connection was terminated")
+			}
+			// a broken stream must fail fast instead of touching the codec again
+			if _, err := client.recv(make([]byte, 64)); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("recv after break = %v, want ErrCodecStreamBroken", err)
+			}
+			if _, err := client.Send([]byte("nope")); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("Send after break = %v, want ErrCodecStreamBroken", err)
+			}
+		})
+	}
+}
+
+// TestCodecSendStalledPeerBreaksStream covers the write half of the P1: a peer
+// that stops reading must not block Send forever. net.Pipe is unbuffered, so a
+// write blocks until the peer reads - exactly the stalled-reader condition.
+func TestCodecSendStalledPeerBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(200 * time.Millisecond)
+			client.SetCompressType(test.compress)
+
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send([]byte("peer never reads this"))
+				sendErr <- err
+			}()
+
+			var err error
+			select {
+			case err = <-sendErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("Send never returned: the write deadline did not reach the socket")
+			}
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+			// the codec connection must be terminated: the stalled peer has to
+			// observe the close instead of a still-open pipe.
+			if err := client.conn.SetWriteDeadline(time.Now()); err == nil {
+				t.Fatal("underlying conn is still open after a stalled write")
+			}
+			if _, err := peerRaw.Write([]byte("ping")); err == nil {
+				t.Fatal("peer write still succeeded after the codec connection was terminated")
+			}
+
+			// the second Send must fail fast rather than block on the dead codec
+			start := time.Now()
+			if _, err = client.Send([]byte("still nope")); !errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("Send after break = %v, want ErrCodecStreamBroken", err)
+			}
+			if elapsed := time.Since(start); elapsed > 100*time.Millisecond {
+				t.Fatalf("Send after break blocked for %s, want an immediate failure", elapsed)
+			}
+		})
+	}
+}
+
+// TestCodecRecvIdleThenResume is the regression guard for the deadline value: a
+// codec stream must survive an idle period of many read timeouts. The read
+// timeout is a poll interval (session.handleTCPPackage retries it), so arming it
+// on a codec stream would kill every idle compressed connection - flate/snappy
+// latch the timeout and never decode again.
+func TestCodecRecvIdleThenResume(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			server.SetWriteTimeout(time.Second)
+			client.SetCompressType(test.compress)
+			server.SetCompressType(test.compress)
+			client.codecStallTimeout = 2 * time.Second
+
+			const idle = 200 * time.Millisecond // 10 read timeouts
+			payload := []byte("packet-after-idle")
+			writeErr := make(chan error, 1)
+			go func() {
+				time.Sleep(idle)
+				_, err := server.Send(payload)
+				writeErr <- err
+			}()
+
+			got := make([]byte, 0, len(payload))
+			buf := make([]byte, 64)
+			for len(got) < len(payload) {
+				n, err := client.recv(buf)
+				if err != nil {
+					t.Fatalf("recv after %s of idling failed: %v", idle, err)
+				}
+				got = append(got, buf[:n]...)
+			}
+			if err := <-writeErr; err != nil {
+				t.Fatalf("peer Send failed: %v", err)
+			}
+			if !bytes.Equal(got, payload) {
+				t.Fatalf("decoded %q, want %q", got, payload)
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("an idle codec stream was wrongly latched as broken")
+			}
+		})
+	}
+}
+
+// TestRawConnRecvTimeoutStaysRetryable pins the other half of the contract: on a
+// raw connection a read timeout is still a benign, retryable net.Error - it is
+// the poll that lets session.handleTCPPackage notice a closed session.
+func TestRawConnRecvTimeoutStaysRetryable(t *testing.T) {
+	client, _ := newTCPConnPair(t)
+	client.SetReadTimeout(50 * time.Millisecond)
+
+	_, err := client.recv(make([]byte, 64))
+	if err == nil {
+		t.Fatal("recv on a silent peer returned no error")
+	}
+	if errors.Is(err, ErrCodecStreamBroken) {
+		t.Fatalf("raw conn read timeout reported as a broken codec stream: %v", err)
+	}
+	netErr, ok := perrors.Cause(err).(net.Error)
+	if !ok || !netErr.Timeout() {
+		t.Fatalf("cause = %v, want a net.Error timeout", perrors.Cause(err))
+	}
+	if client.codecBroken.Load() {
+		t.Fatal("raw conn was latched as broken")
+	}
+}
+
+// TestCodecRecvPeerCloseKeepsErrorIdentity pins the deliberate scope of the
+// latch: only a timeout means "stalled mid-stream". A clean peer close must keep
+// its own EOF-family error, which session.handleTCPPackage matches on to skip
+// reconnecting, and must not latch the connection - otherwise a concurrent
+// WritePkg would report ErrCodecStreamBroken instead of the real cause.
+func TestCodecRecvPeerCloseKeepsErrorIdentity(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(time.Second)
+			client.SetCompressType(test.compress)
+			server.CloseConn(0)
+
+			_, err := client.recv(make([]byte, 64))
+			if err == nil {
+				t.Fatal("recv on a closed peer returned no error")
+			}
+			if errors.Is(err, ErrCodecStreamBroken) {
+				t.Fatalf("clean peer close reported as a stalled codec stream: %v", err)
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("clean peer close latched the connection as broken")
+			}
+		})
+	}
+}

From 3f8fe744ac39cb7efecd51170013014755ab28ae Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Mon, 17 Aug 2026 19:13:16 +0800
Subject: [PATCH 05/15] fix: freeze codec config after first IO and make conn
 close idempotent

SetCompressType raced with concurrent Send/recv on reader/writer, and even a
serialized mid-stream switch would desynchronize the peer's decoder since there
is no codec renegotiation. Guard the codec fields with a mutex and freeze them
once the first recv/Send marks the stream as started; a late SetCompressType
now panics (logged and documented), matching how the method already reports an
illegal compress type. recv/Send snapshot the codec state under the lock and
never hold it across blocking IO.

CloseConn used to nil t.conn while codecIOError/recv/Send read it from other
goroutines. Keep the conn reference immutable and make closing idempotent via
sync.Once instead, for the UDP conn as well.

Add race regression tests: a late SetCompressType must panic on a started
stream, and SetCompressType racing with Send must be rejected without a data
race while the raw stream stays intact.
---
 transport/client_test.go     |  8 ++--
 transport/connection.go      | 90 +++++++++++++++++++++++++-----------
 transport/connection_test.go | 89 +++++++++++++++++++++++++++++++++++
 3 files changed, 156 insertions(+), 31 deletions(-)

diff --git a/transport/client_test.go b/transport/client_test.go
index cb38f096..26bcd8d2 100644
--- a/transport/client_test.go
+++ b/transport/client_test.go
@@ -300,7 +300,9 @@ func TestTCPClient(t *testing.T) {
 	assert.Nil(t, err)
 	active := ss.GetActive()
 	assert.NotNil(t, active)
-	ss.SetCompressType(CompressNone)
+	// the session read loop already started IO, so reconfiguring the codec
+	// mid-stream must be rejected (see SetCompressType)
+	assert.Panics(t, func() { ss.SetCompressType(CompressNone) })
 	conn := ss.(*session).Connection.(*gettyTCPConn)
 	assert.True(t, conn.compress == CompressNone)
 	beforeWriteBytes := conn.writeBytes
@@ -328,7 +330,7 @@ func TestTCPClient(t *testing.T) {
 	beforeWriteBytes.Add(10)
 	assert.Equal(t, beforeWritePkgNum, conn.writePkgNum)
 	assert.Equal(t, beforeWriteBytes, conn.writeBytes)
-	ss.SetCompressType(CompressSnappy)
+	assert.Panics(t, func() { ss.SetCompressType(CompressSnappy) })
 	var anotherPkgs [][]byte
 	anotherPkgs = append(anotherPkgs, []byte("hello"), []byte("hello"))
 	l, err = ss.WriteBytesArray(anotherPkgs...)
@@ -338,7 +340,7 @@ func TestTCPClient(t *testing.T) {
 	beforeWriteBytes.Add(10)
 	assert.Equal(t, beforeWritePkgNum, conn.writePkgNum)
 	assert.Equal(t, beforeWriteBytes, conn.writeBytes)
-	assert.True(t, conn.compress == CompressSnappy)
+	assert.True(t, conn.compress == CompressNone)
 
 	batchSize := 128 * 1023
 	source := make([]byte, batchSize)
diff --git a/transport/connection.go b/transport/connection.go
index a3dcc035..96e82735 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -78,6 +78,9 @@ var CodecStallTimeout = 5 * time.Minute
 // Connection wrap some connection params and operations
 type Connection interface {
 	ID() uint32
+	// SetCompressType sets the compress type. It must be called before any
+	// recv/Send on the connection, typically in NewSessionCallback; a TCP
+	// connection panics if it is called after IO started.
 	SetCompressType(CompressType)
 	LocalAddr() string
 	RemoteAddr() string
@@ -210,9 +213,14 @@ func (c *gettyConn) SetWriteTimeout(wTimeout time.Duration) {
 
 type gettyTCPConn struct {
 	gettyConn
-	reader io.Reader
-	writer io.Writer
-	conn   net.Conn
+	// lock guards the codec fields below; streamStarted is set by the first
+	// recv/Send and freezes the codec configuration (see SetCompressType).
+	lock          sync.Mutex
+	streamStarted bool
+	reader        io.Reader
+	writer        io.Writer
+	conn          net.Conn // immutable after construction, closed via closeOnce
+	closeOnce     sync.Once
 	// codecBroken is set once a codec read/write failed in the middle of the
 	// stream. Written by the read goroutine and read by writer goroutines, so
 	// it has to be atomic. Once set, recv/Send fail fast: touching the codec
@@ -351,8 +359,18 @@ func (s *snappyWriteFlusher) Close() error {
 	return perrors.WithStack(s.writer.Close())
 }
 
-// SetCompressType set compress type(tcp: zip/snappy, websocket:zip)
+// SetCompressType set compress type(tcp: zip/snappy, websocket:zip).
+// It must be called before the first recv/Send on the connection, typically
+// inside NewSessionCallback; there is no codec renegotiation protocol, so
+// switching the stream format after IO started would desynchronize the peer's
+// decoder. Calling it on a started stream panics.
 func (t *gettyTCPConn) SetCompressType(c CompressType) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	if t.streamStarted {
+		log.Errorf("SetCompressType(%d) called after IO started on connection{local:%s, peer:%s}", c, t.local, t.peer)
+		panic("SetCompressType must be called before any recv/Send on the connection, e.g. in NewSessionCallback")
+	}
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
 		ioReader := io.Reader(t.conn)
@@ -408,6 +426,23 @@ func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
 	return timeout
 }
 
+// beginRecv/beginSend mark the stream as started and snapshot the codec state,
+// so the blocking IO below runs without the lock and cannot race with
+// SetCompressType.
+func (t *gettyTCPConn) beginRecv() (io.Reader, time.Duration) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.streamStarted = true
+	return t.reader, t.readDeadlineTimeout()
+}
+
+func (t *gettyTCPConn) beginSend() (io.Writer, bool) {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	t.streamStarted = true
+	return t.writer, t.codecEnabled
+}
+
 // codecIOError maps a timeout on a codec stream to the fatal
 // ErrCodecStreamBroken and latches the connection as broken.
 //
@@ -422,9 +457,8 @@ func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
 // than left half-alive: it unblocks any concurrent Read/Write and lets the
 // session tear down even when the stalled side is the writer. CloseConn is not
 // used here because closing the snappy writer can block on the very peer that
-// stalled; closing the raw conn is enough. t.conn is left in place (session.gc
-// nils it via CloseConn) so a concurrent recv/Send never sees a nil conn, and
-// CloseConn skips the broken writer when it eventually runs.
+// stalled; closing the raw conn is enough, and CloseConn skips the broken
+// writer when it eventually runs.
 //
 // Any other failure keeps its own identity - io.EOF above all, which the session
 // read loop matches on to detect a clean peer shutdown. Those need no latch:
@@ -436,9 +470,7 @@ func (t *gettyTCPConn) codecIOError(err error) error {
 	}
 
 	t.codecBroken.Store(true)
-	if conn := t.conn; conn != nil {
-		_ = conn.Close()
-	}
+	_ = t.conn.Close()
 	return perrors.Wrapf(ErrCodecStreamBroken, "codec stream stalled: %v", err)
 }
 
@@ -483,8 +515,10 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 		return 0, perrors.WithStack(ErrCodecStreamBroken)
 	}
 
+	reader, timeout := t.beginRecv()
+
 	// set read timeout deadline
-	if timeout := t.readDeadlineTimeout(); timeout > 0 {
+	if timeout > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
@@ -495,7 +529,7 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 		t.rLastDeadline.Store(currentTime)
 	}
 
-	length, err = t.reader.Read(p)
+	length, err = reader.Read(p)
 	t.readBytes.Add(uint32(length))
 	return length, t.codecReadError(err)
 }
@@ -515,6 +549,8 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		return 0, perrors.WithStack(ErrCodecStreamBroken)
 	}
 
+	writer, codecEnabled := t.beginSend()
+
 	// The write deadline applies to a codec stream too: unlike a read, a write
 	// only blocks when the peer stopped reading, so it never fires on an idle
 	// connection and there is nothing to poll for. Skipping it here used to make
@@ -534,16 +570,16 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 		// #102: when a codec is installed the [][]byte path must go through
 		// t.writer (the codec writer), otherwise it writes raw frames directly
 		// to t.conn and the peer receives a corrupt mix of coded and raw data.
-		if !t.codecEnabled {
+		if !codecEnabled {
 			// only a raw conn here, so writev the whole batch in one syscall.
 			netBuf := net.Buffers(buffers)
 			lg, err = netBuf.WriteTo(t.conn)
-		} else if bw, ok := t.writer.(buffersWriter); ok {
+		} else if bw, ok := writer.(buffersWriter); ok {
 			lg, err = bw.WriteBuffers(buffers)
 		} else {
 			for _, b := range buffers {
 				var n int
-				n, err = t.writer.Write(b)
+				n, err = writer.Write(b)
 				if err != nil {
 					break
 				}
@@ -560,7 +596,7 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 	}
 
 	if p, ok = pkg.([]byte); ok {
-		length, err = t.writer.Write(p)
+		length, err = writer.Write(p)
 		if err == nil {
 			t.writeBytes.Add((uint32)(len(p)))
 			t.writePkgNum.Add(1)
@@ -575,17 +611,16 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 
 // close tcp connection
 func (t *gettyTCPConn) CloseConn(waitSec int) {
-	// if tcpConn, ok := t.conn.(*net.TCPConn); ok {
-	// tcpConn.SetLinger(0)
-	// }
-
-	if t.conn != nil {
+	t.closeOnce.Do(func() {
+		t.lock.Lock()
+		writer := t.writer
+		t.lock.Unlock()
 		// #102: snappy writer is now wrapped in *snappyWriteFlusher.
 		// A broken codec must not be flushed: the stream is already
 		// desynchronized, and Close would only push more garbage into a socket
 		// that may still be stalled.
 		if !t.codecBroken.Load() {
-			if writer, ok := t.writer.(*snappyWriteFlusher); ok {
+			if writer, ok := writer.(*snappyWriteFlusher); ok {
 				if err := writer.Close(); err != nil {
 					log.Errorf("snappy.Writer.Close() = error:%+v", err)
 				}
@@ -601,8 +636,7 @@ func (t *gettyTCPConn) CloseConn(waitSec int) {
 		} else {
 			_ = t.conn.Close()
 		}
-		t.conn = nil
-	}
+	})
 }
 
 // ///////////////////////////////////////
@@ -621,7 +655,8 @@ func (c UDPContext) String() string {
 type gettyUDPConn struct {
 	gettyConn
 	compressType CompressType
-	conn         *net.UDPConn // for server
+	conn         *net.UDPConn // for server; immutable after construction, closed via closeOnce
+	closeOnce    sync.Once
 }
 
 // create gettyUDPConn
@@ -730,10 +765,9 @@ func (u *gettyUDPConn) Send(udpCtx any) (int, error) {
 
 // close udp connection
 func (u *gettyUDPConn) CloseConn(_ int) {
-	if u.conn != nil {
+	u.closeOnce.Do(func() {
 		_ = u.conn.Close()
-		u.conn = nil
-	}
+	})
 }
 
 // ///////////////////////////////////////
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 8b0ba67a..c045235b 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -592,3 +592,92 @@ func TestCodecRecvPeerCloseKeepsErrorIdentity(t *testing.T) {
 		})
 	}
 }
+
+// TestSetCompressTypeAfterIORejected pins the codec configuration contract:
+// once a connection has sent or received, SetCompressType must panic instead
+// of replacing the codec under in-flight IO and desynchronizing the peer.
+func TestSetCompressTypeAfterIORejected(t *testing.T) {
+	client, server := newTCPConnPair(t)
+
+	payload := []byte("started")
+	gotCh, errCh := startReceiver(server, len(payload), 8)
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+	select {
+	case <-gotCh:
+	case err := <-errCh:
+		t.Fatalf("peer failed to read: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the payload")
+	}
+
+	for name, conn := range map[string]*gettyTCPConn{"sender": client, "receiver": server} {
+		func() {
+			defer func() {
+				if recover() == nil {
+					t.Fatalf("SetCompressType on a started %s did not panic", name)
+				}
+			}()
+			conn.SetCompressType(CompressSnappy)
+		}()
+	}
+}
+
+// TestSetCompressTypeConcurrentWithSend is the race regression for the PR#107
+// review: SetCompressType(CompressSnappy) racing with Send([]byte) on a started
+// stream must be rejected, stay race-free under `go test -race` and leave the
+// raw stream intact.
+func TestSetCompressTypeConcurrentWithSend(t *testing.T) {
+	client, server := newTCPConnPair(t)
+
+	payload := []byte("payload-")
+	const sends = 100
+	expected := bytes.Repeat(payload, sends+1)
+	gotCh, errCh := startReceiver(server, len(expected), 16)
+
+	// start the stream, so the SetCompressType below is guaranteed to be late
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+
+	start := make(chan struct{})
+	panicked := make(chan bool, 1)
+	var wg sync.WaitGroup
+	wg.Add(2)
+	go func() {
+		defer wg.Done()
+		<-start
+		for i := 0; i < sends; i++ {
+			if _, err := client.Send(payload); err != nil {
+				t.Errorf("Send failed: %v", err)
+				return
+			}
+		}
+	}()
+	go func() {
+		defer wg.Done()
+		defer func() { panicked <- recover() != nil }()
+		<-start
+		client.SetCompressType(CompressSnappy)
+	}()
+	close(start)
+	wg.Wait()
+
+	if !<-panicked {
+		t.Fatal("late SetCompressType was not rejected")
+	}
+	if client.codecEnabled {
+		t.Fatal("rejected SetCompressType still installed a codec")
+	}
+	select {
+	case got := <-gotCh:
+		if !bytes.Equal(got, expected) {
+			t.Fatalf("stream corrupted:\n got: %q\nwant: %q", got, expected)
+		}
+	case err := <-errCh:
+		t.Fatalf("peer failed to read: %v", err)
+	case <-time.After(10 * time.Second):
+		t.Fatal("timed out waiting for the stream")
+	}
+}

From 18b76ebf6bd5ae004b887fd4e39482e4cd2fdb25 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Mon, 17 Aug 2026 22:47:59 +0800
Subject: [PATCH 06/15] fix: close TestTCPClient listener on cleanup and
 correct batch-send debug log

TestTCPClient never closed its listener, leaving the accept goroutine
blocked in Accept() for the rest of the test binary (review P2 by
@AlexStocks). The [][]byte send path logged the never-assigned length
variable instead of the actual written byte count and formatted a nil
error with %s (flagged by copilot review).
---
 transport/client_test.go | 1 +
 transport/connection.go  | 4 ++--
 2 files changed, 3 insertions(+), 2 deletions(-)

diff --git a/transport/client_test.go b/transport/client_test.go
index 26bcd8d2..1cfe847a 100644
--- a/transport/client_test.go
+++ b/transport/client_test.go
@@ -272,6 +272,7 @@ func TestTCPClient(t *testing.T) {
 	listener, err := listenLocalServer()
 	assert.Nil(t, err)
 	assert.NotNil(t, listener)
+	t.Cleanup(func() { _ = listener.Close() })
 
 	addr := listener.Addr().(*net.TCPAddr)
 	t.Logf("server addr: %v", addr)
diff --git a/transport/connection.go b/transport/connection.go
index 96e82735..6ebd3ff0 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -590,8 +590,8 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 			t.writeBytes.Add((uint32)(lg))
 			t.writePkgNum.Add((uint32)(len(buffers)))
 		}
-		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%s",
-			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
+		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%v",
+			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, lg, err)
 		return int(lg), t.codecIOError(err)
 	}
 

From 07be3738eb20c513aee62a987b90b570c0757274 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 01:58:30 +0800
Subject: [PATCH 07/15] fix: only treat mid-block silence as a codec stall,
 never pure idleness

CodecStallTimeout was armed as a flat read deadline on every codec stream
and codecIOError declared any timeout fatal, so a compressed connection
that simply stayed idle longer than the timeout (default 5min) was falsely
latched as broken and its socket closed - reviewed as P1: the deadline
could not tell 'received half a codec block then silence' from 'never
received a byte at all'.

Install codecPollingReader between the codec and the raw conn. It owns the
read deadlines, polling with rTimeout like a raw connection, and classifies
each timeout where the information exists: idle timeouts (no bytes since
the last fully decoded block) are absorbed and never reach the
error-latching flate/snappy readers; only silence of at least
codecStallTimeout after partial codec data arrived (progress since the
decode boundary recv reports back) surfaces to the codec and breaks the
connection. Session shutdown wakeups pass through via the existing
sessionClosing exemption. recv no longer arms deadlines on codec streams.

Known bound, documented on the type: a block spanning two recv calls whose
remainder is already buffered inside the decoder is indistinguishable from
idleness at this layer and waits for session close instead of breaking early.

Add TestCodecRecvIdleBeyondStallTimeoutStaysHealthy pinning the reviewer's
probe: zero-byte silence and between-packet idleness at 4x the stall
timeout must survive and resume; the existing mid-block stall tests still
pass unchanged.
---
 transport/connection.go      | 162 ++++++++++++++++++++++++++---------
 transport/connection_test.go |  64 ++++++++++++++
 2 files changed, 186 insertions(+), 40 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 6ebd3ff0..8ac6d6e4 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -57,22 +57,24 @@ var (
 // instead of as a benign read timeout that can be retried.
 var ErrCodecStreamBroken = errors.New("getty: codec stream is broken, connection must be closed")
 
-// CodecStallTimeout is the read deadline used on a codec(compressed) stream,
-// and thus the upper bound on how long a stalled peer can block a read.
+// CodecStallTimeout bounds how long a codec(compressed) stream may stay silent
+// AFTER it has delivered part of a codec block: that is the only situation
+// where the peer provably died mid-write and the decoder state is
+// unrecoverable. A stream that is merely idle - no bytes at all since the last
+// fully decoded block - is healthy and is never subject to this timeout, no
+// matter how long the silence lasts.
 //
-// It exists because the connection read timeout (see SetReadTimeout, 1s by
-// default) is a poll interval, not an idle timeout: session.handleTCPPackage
-// retries after every read timeout. A codec stream cannot be retried
-// (flate/snappy latch read errors forever), so arming rTimeout on it would tear
-// down every compressed connection that stays idle for one second. Hence codec
-// reads get this much wider deadline instead, and hitting it means the peer
-// really stalled mid-stream: the connection is declared broken.
+// The mechanics live in codecPollingReader: the connection read timeout (see
+// SetReadTimeout, 1s by default) is used as a poll interval underneath the
+// codec, and poll timeouts are absorbed there instead of reaching the
+// flate/snappy reader (which would latch any error forever). Only a
+// mid-block stall of at least CodecStallTimeout is surfaced to the codec,
+// which is what declares the connection broken.
 //
-// Keep it well above the read timeout and above the application heartbeat
-// period, otherwise idle compressed connections get killed. Set it to 0 to arm
-// no read deadline at all on codec streams (a stalled peer then blocks until
-// the session is closed). It is read once, when SetCompressType installs the
-// codec, so set it before creating connections.
+// Set it to 0 to disable stall detection (a peer that dies mid-block then
+// blocks the read until the session is closed). It is copied per connection
+// when SetCompressType installs the codec, so set it before creating
+// connections.
 var CodecStallTimeout = 5 * time.Minute
 
 // Connection wrap some connection params and operations
@@ -227,8 +229,11 @@ type gettyTCPConn struct {
 	// again would only produce more garbage on the wire.
 	codecBroken uatomic.Bool
 	// codecStallTimeout is the per-conn copy of CodecStallTimeout, taken when
-	// the codec is installed. 0 arms no read deadline on the codec stream.
+	// the codec is installed. 0 disables mid-block stall detection.
 	codecStallTimeout time.Duration
+	// pollReader is the deadline-owning reader installed between the codec and
+	// the raw conn by SetCompressType; nil on a raw connection.
+	pollReader *codecPollingReader
 }
 
 // create gettyTCPConn
@@ -371,10 +376,14 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 		log.Errorf("SetCompressType(%d) called after IO started on connection{local:%s, peer:%s}", c, t.local, t.peer)
 		panic("SetCompressType must be called before any recv/Send on the connection, e.g. in NewSessionCallback")
 	}
+	// The codec never reads the raw conn directly: codecPollingReader sits in
+	// between, owns the read deadlines and absorbs idle poll timeouts, so the
+	// error-latching flate/snappy readers only ever see a genuine mid-block
+	// stall (or a real IO error).
+	poller := &codecPollingReader{t: t}
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
-		ioReader := io.Reader(t.conn)
-		t.reader = flate.NewReader(ioReader)
+		t.reader = flate.NewReader(poller)
 
 		ioWriter := io.Writer(t.conn)
 		w, err := flate.NewWriter(ioWriter, int(c))
@@ -384,8 +393,7 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 		t.writer = &writeFlusher{flusher: w}
 
 	case CompressSnappy:
-		ioReader := io.Reader(t.conn)
-		t.reader = snappy.NewReader(ioReader)
+		t.reader = snappy.NewReader(poller)
 		ioWriter := io.Writer(t.conn)
 		// #102: wrap the buffered snappy writer so every Write is flushed,
 		// otherwise small packets never leave the internal buffer.
@@ -394,6 +402,7 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
+	t.pollReader = poller
 	// Both branches replaced reader/writer with a codec (CompressNone included,
 	// see the codecEnabled comment), so the conn is no longer raw: all IO must
 	// go through t.reader/t.writer, and a read/write error can no longer be
@@ -403,37 +412,104 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 	t.compress = c
 }
 
-// readDeadlineTimeout returns the deadline to arm before a read.
+// codecPollingReader is the io.Reader the codec (flate/snappy) reads from
+// instead of the raw conn. It exists because those decoders latch the first
+// error their source ever returns, so a plain read timeout must never reach
+// them - yet without timeouts a dead peer blocks the read goroutine forever.
 //
-// A raw conn uses rTimeout, whose timeouts session.handleTCPPackage simply
-// retries. A codec stream cannot be retried, so it gets CodecStallTimeout
-// instead - see that variable for why rTimeout is unusable here. The larger of
-// the two wins: raising rTimeout widens the codec bound, and lowering
-// CodecStallTimeout tightens it.
-func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
-	timeout := t.rTimeout.Load()
-	if !t.codecEnabled {
-		return timeout
+// The reader polls the conn with rTimeout deadlines (the same poll interval a
+// raw connection uses) and classifies each timeout:
+//   - no bytes seen since the last fully decoded block: the stream is idle,
+//     which is healthy; the timeout is absorbed and the poll continues.
+//   - partial codec data seen (progress) and silence for at least
+//     codecStallTimeout: the peer died mid-block, the decoder state is
+//     unrecoverable; the timeout is surfaced, latched by the codec and turned
+//     into ErrCodecStreamBroken by codecReadError.
+//
+// Known bound: if a block spans two recv calls and the tail silence starts
+// after the decoder buffered the partial remainder internally, no wrapper-level
+// progress exists and the stall is indistinguishable from idle; such a
+// connection lives until the session is closed instead of being broken early.
+//
+// All fields are only touched by the single read goroutine; session.stop()
+// interacts with it solely by arming a conn deadline (the wakeup is seen here
+// as a timeout and passed through once sessionClosing()/codecBroken is set).
+type codecPollingReader struct {
+	t *gettyTCPConn
+	// progress is true once raw bytes arrived after the last decode boundary
+	// (see boundary), i.e. the decoder may be holding part of a codec block.
+	progress bool
+	// lastByte is when the most recent raw byte arrived.
+	lastByte time.Time
+}
+
+// boundary is called by recv after the codec returned decoded output: the
+// stream is at a block boundary again, so subsequent silence is idleness, not
+// a stall.
+func (r *codecPollingReader) boundary() {
+	r.progress = false
+}
+
+func (r *codecPollingReader) Read(p []byte) (int, error) {
+	t := r.t
+	for {
+		if timeout := t.rTimeout.Load(); timeout > 0 {
+			// Set Deadline every time, since golang has fixed the performance issue
+			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
+			currentTime := time.Now()
+			if err := t.conn.SetReadDeadline(currentTime.Add(timeout)); err != nil {
+				return 0, err
+			}
+			t.rLastDeadline.Store(currentTime)
+		}
+
+		n, err := t.conn.Read(p)
+		if n > 0 {
+			r.progress = true
+			r.lastByte = time.Now()
+		}
+		if err == nil || !isTimeoutError(err) {
+			// real data or a real error: both keep their identity.
+			return n, err
+		}
+		if n > 0 {
+			// data arrived together with the deadline: deliver it and swallow
+			// the timeout, the codec will come back for more.
+			return n, nil
+		}
+		if t.codecBroken.Load() || t.sessionClosing() {
+			// shutdown wakeup (session.stop arms a deadline for exactly this),
+			// codecReadError passes it through as a plain timeout.
+			return 0, err
+		}
+		if stall := t.codecStallTimeout; r.progress && stall > 0 && time.Since(r.lastByte) >= stall {
+			// mid-block stall: surface the timeout, the codec latches it and
+			// codecReadError declares the stream broken.
+			return 0, err
+		}
+		// idle poll timeout: absorb and keep waiting.
 	}
-	if t.codecStallTimeout <= 0 {
-		// stall cap disabled: no deadline, the read blocks until data arrives
-		// or session.stop() arms a deadline to unblock it.
+}
+
+// readDeadlineTimeout returns the deadline recv arms before a read. Only a raw
+// conn needs one (session.handleTCPPackage retries its timeouts as a poll); on
+// a codec stream the deadlines are owned by codecPollingReader underneath the
+// decoder, so recv itself must not arm any.
+func (t *gettyTCPConn) readDeadlineTimeout() time.Duration {
+	if t.codecEnabled {
 		return 0
 	}
-	if timeout < t.codecStallTimeout {
-		return t.codecStallTimeout
-	}
-	return timeout
+	return t.rTimeout.Load()
 }
 
 // beginRecv/beginSend mark the stream as started and snapshot the codec state,
 // so the blocking IO below runs without the lock and cannot race with
 // SetCompressType.
-func (t *gettyTCPConn) beginRecv() (io.Reader, time.Duration) {
+func (t *gettyTCPConn) beginRecv() (io.Reader, *codecPollingReader, time.Duration) {
 	t.lock.Lock()
 	defer t.lock.Unlock()
 	t.streamStarted = true
-	return t.reader, t.readDeadlineTimeout()
+	return t.reader, t.pollReader, t.readDeadlineTimeout()
 }
 
 func (t *gettyTCPConn) beginSend() (io.Writer, bool) {
@@ -515,9 +591,10 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 		return 0, perrors.WithStack(ErrCodecStreamBroken)
 	}
 
-	reader, timeout := t.beginRecv()
+	reader, poller, timeout := t.beginRecv()
 
-	// set read timeout deadline
+	// set read timeout deadline (raw conn only; a codec stream's deadlines are
+	// owned by codecPollingReader, see readDeadlineTimeout)
 	if timeout > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
@@ -530,6 +607,11 @@ func (t *gettyTCPConn) recv(p []byte) (int, error) {
 	}
 
 	length, err = reader.Read(p)
+	if poller != nil && length > 0 && err == nil {
+		// decoded output was produced: the stream is at a block boundary,
+		// silence from here on is idleness rather than a mid-block stall.
+		poller.boundary()
+	}
 	t.readBytes.Add(uint32(length))
 	return length, t.codecReadError(err)
 }
diff --git a/transport/connection_test.go b/transport/connection_test.go
index c045235b..7370e38e 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -543,6 +543,70 @@ func TestCodecRecvIdleThenResume(t *testing.T) {
 	}
 }
 
+// TestCodecRecvIdleBeyondStallTimeoutStaysHealthy pins the stall/idle
+// distinction: CodecStallTimeout only applies to a stream that stalled in the
+// middle of a codec block. A connection that never received a byte, or that is
+// idle between fully decoded packets, must survive silence far beyond the
+// stall timeout and resume normally - previously any silence longer than the
+// timeout was misclassified as a broken stream and the socket was closed.
+func TestCodecRecvIdleBeyondStallTimeoutStaysHealthy(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 100 * time.Millisecond
+			server.SetWriteTimeout(time.Second)
+			server.SetCompressType(test.compress)
+
+			payload := []byte("packet-after-long-idle")
+			expected := bytes.Repeat(payload, 2)
+			gotCh, errCh := startReceiver(client, len(expected), 8)
+
+			// silence with zero bytes ever received, 4x the stall timeout
+			const idle = 400 * time.Millisecond
+			select {
+			case err := <-errCh:
+				t.Fatalf("connection that never received a byte was killed after idling: %v", err)
+			case <-time.After(idle):
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("never-used codec stream was latched as broken by pure idleness")
+			}
+			if _, err := server.Send(payload); err != nil {
+				t.Fatalf("peer Send after idle failed: %v", err)
+			}
+
+			// idle again between two fully decoded packets, then resume
+			select {
+			case err := <-errCh:
+				t.Fatalf("connection idling between packets was killed: %v", err)
+			case <-time.After(idle):
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("codec stream idling between packets was latched as broken")
+			}
+			if _, err := server.Send(payload); err != nil {
+				t.Fatalf("peer Send after second idle failed: %v", err)
+			}
+
+			select {
+			case got := <-gotCh:
+				if !bytes.Equal(got, expected) {
+					t.Fatalf("decoded stream mismatch:\n got: %q\nwant: %q", got, expected)
+				}
+			case err := <-errCh:
+				t.Fatalf("peer failed to decode after idle periods: %v", err)
+			case <-time.After(10 * time.Second):
+				t.Fatal("timed out waiting for the decoded stream")
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("healthy idle connection ended up latched as broken")
+			}
+		})
+	}
+}
+
 // TestRawConnRecvTimeoutStaysRetryable pins the other half of the contract: on a
 // raw connection a read timeout is still a benign, retryable net.Error - it is
 // the poll that lets session.handleTCPPackage notice a closed session.

From 18a2687249610a030e5ef7044e3e84c7ee91d101 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 02:28:06 +0800
Subject: [PATCH 08/15] fix: detect cross-recv codec stalls exactly and stop
 killing slow-draining peers on write

Two refinements of the stall/idle classification:

Read side: codecPollingReader now implements io.ByteReader and carries the
only read-ahead buffer under the codec. flate uses a Reader+ByteReader source
directly instead of wrapping it in a hidden bufio.Reader, so bytes read ahead
of the current block always live where the classifier can see them. Progress
now means 'bytes delivered to the decoder since the last decode boundary',
which closes the documented blind spot: a block spanning two recv calls whose
remainder was buffered ahead of the peer dying is detected as a mid-block
stall instead of idling until session close. A compile-time flate.Reader
assertion pins the no-hidden-bufio guarantee. snappy reads exactly per chunk
and gains batched refills.

Write side: a hard wTimeout over a whole compressed burst broke connections
whose peer merely drained slowly (one WriteBuffers batch shares a single
deadline). codecPollingWriter now owns the write deadlines, using wTimeout as
a poll interval and resuming partial writes from the exact position - safe
only at this layer, which owns the byte position. A timeout is absorbed while
attempts keep making progress; zero progress for codecStallTimeout, or a
shutdown, surfaces it and breaks the stream as before. Send no longer arms
write deadlines on codec streams.

Regression tests: a full-block-plus-half-block burst followed by silence must
break the stream (previously hung as idleness), and a peer draining 8 bytes
per 30ms must survive a burst far exceeding wTimeout. The existing stall,
idle and shutdown tests pass unchanged.
---
 transport/connection.go      | 174 ++++++++++++++++++++++++++++-------
 transport/connection_test.go | 152 +++++++++++++++++++++++++++++-
 2 files changed, 292 insertions(+), 34 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 8ac6d6e4..37ec9c7c 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -71,6 +71,11 @@ var ErrCodecStreamBroken = errors.New("getty: codec stream is broken, connection
 // mid-block stall of at least CodecStallTimeout is surfaced to the codec,
 // which is what declares the connection broken.
 //
+// The write side (codecPollingWriter) applies the same bound to a peer that
+// stopped reading: wTimeout is the poll interval, and a write attempt that
+// makes no progress for CodecStallTimeout breaks the stream, while a slow but
+// draining peer is waited on indefinitely.
+//
 // Set it to 0 to disable stall detection (a peer that dies mid-block then
 // blocks the read until the session is closed). It is copied per connection
 // when SetCompressType installs the codec, so set it before creating
@@ -380,13 +385,13 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 	// between, owns the read deadlines and absorbs idle poll timeouts, so the
 	// error-latching flate/snappy readers only ever see a genuine mid-block
 	// stall (or a real IO error).
-	poller := &codecPollingReader{t: t}
+	poller := &codecPollingReader{t: t, buf: make([]byte, maxReadBufLen)}
+	pollWriter := &codecPollingWriter{t: t}
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
 		t.reader = flate.NewReader(poller)
 
-		ioWriter := io.Writer(t.conn)
-		w, err := flate.NewWriter(ioWriter, int(c))
+		w, err := flate.NewWriter(pollWriter, int(c))
 		if err != nil {
 			panic(fmt.Sprintf("flate.NewReader(flate.DefaultCompress) = err(%s)", err))
 		}
@@ -394,10 +399,9 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 
 	case CompressSnappy:
 		t.reader = snappy.NewReader(poller)
-		ioWriter := io.Writer(t.conn)
 		// #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))
+		t.writer = newSnappyWriteFlusher(snappy.NewBufferedWriter(pollWriter))
 
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
@@ -426,23 +430,37 @@ func (t *gettyTCPConn) SetCompressType(c CompressType) {
 //     unrecoverable; the timeout is surfaced, latched by the codec and turned
 //     into ErrCodecStreamBroken by codecReadError.
 //
-// Known bound: if a block spans two recv calls and the tail silence starts
-// after the decoder buffered the partial remainder internally, no wrapper-level
-// progress exists and the stall is indistinguishable from idle; such a
-// connection lives until the session is closed instead of being broken early.
+// The reader also implements io.ByteReader and carries the ONLY read-ahead
+// buffer under the codec: flate uses a Reader+ByteReader source directly
+// instead of wrapping it in a hidden bufio.Reader (snappy never reads ahead),
+// so bytes read ahead of the current codec block always live in r.buf where
+// this classification can see them. Progress therefore means "bytes were
+// delivered to the decoder since the last decode boundary" - a partial block
+// that spans two recv calls is still detected as a stall, because its
+// remainder is delivered from r.buf after the boundary.
 //
 // All fields are only touched by the single read goroutine; session.stop()
 // interacts with it solely by arming a conn deadline (the wakeup is seen here
 // as a timeout and passed through once sessionClosing()/codecBroken is set).
 type codecPollingReader struct {
 	t *gettyTCPConn
-	// progress is true once raw bytes arrived after the last decode boundary
-	// (see boundary), i.e. the decoder may be holding part of a codec block.
+	// buf/next/end: read-ahead buffer, next..end is the undelivered remainder.
+	buf  []byte
+	next int
+	end  int
+	// progress is true once bytes were delivered to the decoder after the last
+	// decode boundary (see boundary), i.e. the decoder may be holding part of
+	// a codec block.
 	progress bool
-	// lastByte is when the most recent raw byte arrived.
-	lastByte time.Time
+	// lastDelivery is when the decoder last received bytes.
+	lastDelivery time.Time
 }
 
+// The ByteReader half of this assertion is load-bearing: without it flate
+// silently wraps the poller in its own bufio.Reader and read-ahead residue
+// becomes invisible again (the cross-recv stall blind spot returns).
+var _ flate.Reader = (*codecPollingReader)(nil)
+
 // boundary is called by recv after the codec returned decoded output: the
 // stream is at a block boundary again, so subsequent silence is idleness, not
 // a stall.
@@ -450,7 +468,39 @@ func (r *codecPollingReader) boundary() {
 	r.progress = false
 }
 
+func (r *codecPollingReader) markDelivery() {
+	r.progress = true
+	r.lastDelivery = time.Now()
+}
+
 func (r *codecPollingReader) Read(p []byte) (int, error) {
+	if r.next == r.end {
+		if err := r.fill(); err != nil {
+			return 0, err
+		}
+	}
+	n := copy(p, r.buf[r.next:r.end])
+	r.next += n
+	r.markDelivery()
+	return n, nil
+}
+
+// ReadByte makes this a flate.Reader source: flate then consumes exactly the
+// bytes it needs through here rather than over-reading via its own bufio.
+func (r *codecPollingReader) ReadByte() (byte, error) {
+	if r.next == r.end {
+		if err := r.fill(); err != nil {
+			return 0, err
+		}
+	}
+	b := r.buf[r.next]
+	r.next++
+	r.markDelivery()
+	return b, nil
+}
+
+// fill polls the conn until data arrives or a non-absorbable condition is hit.
+func (r *codecPollingReader) fill() error {
 	t := r.t
 	for {
 		if timeout := t.rTimeout.Load(); timeout > 0 {
@@ -458,39 +508,97 @@ func (r *codecPollingReader) Read(p []byte) (int, error) {
 			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 			currentTime := time.Now()
 			if err := t.conn.SetReadDeadline(currentTime.Add(timeout)); err != nil {
-				return 0, err
+				return err
 			}
 			t.rLastDeadline.Store(currentTime)
 		}
 
-		n, err := t.conn.Read(p)
+		n, err := t.conn.Read(r.buf)
 		if n > 0 {
-			r.progress = true
-			r.lastByte = time.Now()
+			// data first: an error that fired while data was pending (even a
+			// real one) reoccurs on the next conn.Read.
+			r.next, r.end = 0, n
+			return nil
 		}
-		if err == nil || !isTimeoutError(err) {
-			// real data or a real error: both keep their identity.
-			return n, err
+		if err == nil {
+			continue
 		}
-		if n > 0 {
-			// data arrived together with the deadline: deliver it and swallow
-			// the timeout, the codec will come back for more.
-			return n, nil
+		if !isTimeoutError(err) {
+			// a real error keeps its identity - io.EOF above all.
+			return err
 		}
 		if t.codecBroken.Load() || t.sessionClosing() {
 			// shutdown wakeup (session.stop arms a deadline for exactly this),
 			// codecReadError passes it through as a plain timeout.
-			return 0, err
+			return err
 		}
-		if stall := t.codecStallTimeout; r.progress && stall > 0 && time.Since(r.lastByte) >= stall {
+		if stall := t.codecStallTimeout; r.progress && stall > 0 && time.Since(r.lastDelivery) >= stall {
 			// mid-block stall: surface the timeout, the codec latches it and
 			// codecReadError declares the stream broken.
-			return 0, err
+			return err
 		}
 		// idle poll timeout: absorb and keep waiting.
 	}
 }
 
+// codecPollingWriter is the write-side twin of codecPollingReader: the codec
+// writers (flate/snappy) latch the first error just like the readers, and a
+// hard wTimeout over a whole compressed burst would break connections whose
+// peer is merely slow. So write deadlines are owned here, wTimeout acts as a
+// poll interval, and a timeout is absorbed as long as the last attempt made
+// progress recently; only zero progress for codecStallTimeout (a peer that
+// genuinely stopped reading) or a shutdown surfaces the timeout, which the
+// codec latches and codecIOError turns into ErrCodecStreamBroken.
+//
+// A partial conn.Write on timeout is resumed from the exact position, which is
+// only safe because this layer owns the byte position - callers above the
+// codec could never retry a partial write without desynchronizing the stream.
+type codecPollingWriter struct {
+	t *gettyTCPConn
+}
+
+func (w *codecPollingWriter) Write(p []byte) (int, error) {
+	t := w.t
+	written := 0
+	lastProgress := time.Now()
+	for written < len(p) {
+		if timeout := t.wTimeout.Load(); timeout > 0 {
+			// Set Deadline every time, since golang has fixed the performance issue
+			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
+			currentTime := time.Now()
+			if err := t.conn.SetWriteDeadline(currentTime.Add(timeout)); err != nil {
+				return written, err
+			}
+			t.wLastDeadline.Store(currentTime)
+		}
+
+		n, err := t.conn.Write(p[written:])
+		written += n
+		if n > 0 {
+			lastProgress = time.Now()
+		}
+		if err == nil {
+			continue
+		}
+		if !isTimeoutError(err) {
+			return written, err
+		}
+		if t.codecBroken.Load() || t.sessionClosing() {
+			// on shutdown the write is abandoned mid-block; the connection is
+			// being torn down anyway, and latching codecBroken here also stops
+			// CloseConn from flushing the half-written snappy stream.
+			return written, err
+		}
+		if stall := t.codecStallTimeout; stall > 0 && time.Since(lastProgress) >= stall {
+			// the peer accepted nothing for a whole stall window: it stopped
+			// reading for good, surface the timeout and break the stream.
+			return written, err
+		}
+		// slow but progressing peer: absorb the timeout and keep writing.
+	}
+	return written, nil
+}
+
 // readDeadlineTimeout returns the deadline recv arms before a read. Only a raw
 // conn needs one (session.handleTCPPackage retries its timeouts as a poll); on
 // a codec stream the deadlines are owned by codecPollingReader underneath the
@@ -633,12 +741,12 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 
 	writer, codecEnabled := t.beginSend()
 
-	// The write deadline applies to a codec stream too: unlike a read, a write
-	// only blocks when the peer stopped reading, so it never fires on an idle
-	// connection and there is nothing to poll for. Skipping it here used to make
-	// SetWriteTimeout - and the per-call WritePkg(pkg, timeout) - silently
-	// ineffective, letting a stalled peer block writers forever.
-	if t.wTimeout.Load() > 0 {
+	// A raw conn arms the write deadline here, so SetWriteTimeout - and the
+	// per-call WritePkg(pkg, timeout) - actually bounds a write to a stalled
+	// peer. On a codec stream the deadlines are owned by codecPollingWriter
+	// underneath the codec (wTimeout is its poll interval there), which
+	// distinguishes a slow-but-progressing peer from one that stopped reading.
+	if !codecEnabled && t.wTimeout.Load() > 0 {
 		// Set Deadline every time, since golang has fixed the performance issue
 		// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 		currentTime = time.Now()
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 7370e38e..009f24d8 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -21,6 +21,7 @@ import (
 	"bytes"
 	"compress/flate"
 	"errors"
+	"fmt"
 	"io"
 	"net"
 	"sync"
@@ -457,8 +458,10 @@ func TestCodecSendStalledPeerBreaksStream(t *testing.T) {
 			})
 
 			client := newGettyTCPConn(clientRaw)
-			client.SetWriteTimeout(200 * time.Millisecond)
+			client.SetWriteTimeout(50 * time.Millisecond)
 			client.SetCompressType(test.compress)
+			// zero write progress for this long means the peer stopped reading
+			client.codecStallTimeout = 200 * time.Millisecond
 
 			sendErr := make(chan error, 1)
 			go func() {
@@ -607,6 +610,153 @@ func TestCodecRecvIdleBeyondStallTimeoutStaysHealthy(t *testing.T) {
 	}
 }
 
+// TestCodecRecvStallAcrossRecvBoundaryBreaksStream pins the exactness of the
+// stall detection: the peer flushes one complete packet plus the first half of
+// the next block in a single burst, then dies. The remainder is delivered to
+// the decoder from the poller's own read-ahead buffer after the first packet
+// decoded, so the following silence must still be classified as a mid-block
+// stall - with a hidden bufio between poller and decoder this case was
+// indistinguishable from idleness and hung until session close.
+func TestCodecRecvStallAcrossRecvBoundaryBreaksStream(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			// one codec stream: full flushed block for payload1, then payload2's
+			// block truncated in half - exactly what a peer that dies mid-write
+			// leaves after a healthy packet.
+			payload1 := []byte("complete-first-packet")
+			var (
+				buf    bytes.Buffer
+				writer interface {
+					io.Writer
+					Flush() error
+				}
+			)
+			if test.compress == CompressSnappy {
+				writer = snappy.NewBufferedWriter(&buf)
+			} else {
+				flateWriter, err := flate.NewWriter(&buf, int(test.compress))
+				if err != nil {
+					t.Fatal(err)
+				}
+				writer = flateWriter
+			}
+			if _, err := writer.Write(payload1); err != nil {
+				t.Fatal(err)
+			}
+			if err := writer.Flush(); err != nil {
+				t.Fatal(err)
+			}
+			firstLen := buf.Len()
+			if _, err := writer.Write([]byte("second-packet-that-never-finishes")); err != nil {
+				t.Fatal(err)
+			}
+			if err := writer.Flush(); err != nil {
+				t.Fatal(err)
+			}
+			second := buf.Bytes()[firstLen:]
+			if len(second) < 4 {
+				t.Fatalf("second codec block too small to truncate: %d bytes", len(second))
+			}
+			burst := buf.Bytes()[:firstLen+len(second)/2]
+
+			if _, err := server.conn.Write(burst); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+			// the peer now dies: no more bytes, connection stays open.
+
+			recvErr := make(chan error, 1)
+			go func() {
+				var got []byte
+				buf := make([]byte, 64)
+				for {
+					n, err := client.recv(buf)
+					if err != nil {
+						recvErr <- err
+						return
+					}
+					got = append(got, buf[:n]...)
+					if len(got) > len(payload1) {
+						recvErr <- fmt.Errorf("decoded beyond the first packet: %q", got)
+						return
+					}
+				}
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("recv never returned: the cross-recv stall was classified as idleness")
+			}
+			assertCodecStreamBroken(t, err)
+			if !client.codecBroken.Load() {
+				t.Fatal("connection was not latched as broken")
+			}
+		})
+	}
+}
+
+// TestCodecSendSlowPeerBeyondWriteTimeoutSurvives is the write-side twin of the
+// idle/stall distinction: a peer that drains slowly but steadily must not be
+// killed just because one compressed burst takes longer than wTimeout - only
+// zero progress for codecStallTimeout may break the stream.
+func TestCodecSendSlowPeerBeyondWriteTimeoutSurvives(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 500 * time.Millisecond
+
+			// the peer drains a few bytes at a time, far slower than wTimeout
+			// allows for the whole burst, but never stops for a stall window.
+			peerDone := make(chan struct{})
+			go func() {
+				defer close(peerDone)
+				buf := make([]byte, 8)
+				for {
+					if _, err := peerRaw.Read(buf); err != nil {
+						return
+					}
+					time.Sleep(30 * time.Millisecond)
+				}
+			}()
+
+			payload := bytes.Repeat([]byte("slow-but-alive-"), 20) // 300 bytes
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send(payload)
+				sendErr <- err
+			}()
+
+			select {
+			case err := <-sendErr:
+				if err != nil {
+					t.Fatalf("Send to a slow but draining peer failed: %v", err)
+				}
+			case <-time.After(10 * time.Second):
+				t.Fatal("Send never completed against a slow but draining peer")
+			}
+			if client.codecBroken.Load() {
+				t.Fatal("slow but draining peer was latched as a broken stream")
+			}
+			_ = clientRaw.Close()
+			<-peerDone
+		})
+	}
+}
+
 // TestRawConnRecvTimeoutStaysRetryable pins the other half of the contract: on a
 // raw connection a read timeout is still a benign, retryable net.Error - it is
 // the poll that lets session.handleTCPPackage notice a closed session.

From 15a0a26a8ed984269aa217052c91ea1e80bfd473 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 02:31:43 +0800
Subject: [PATCH 09/15] fix: freeze websocket compression config after first
 IO, matching the TCP contract

gettyWSConn.SetCompressType called gorilla's EnableWriteCompression and
SetCompressionLevel - plain field writes - with no synchronization against
in-flight writers (Send, heartbeat writePing), a data race under a running
session. Apply the same contract the TCP connection got earlier in this
branch: compression must be configured before the first recv/Send (i.e. in
NewSessionCallback) and a late call panics. The check runs under writeLock so
it serializes race-free against writers; readers/writers mark the stream as
started inside their existing locks.

TestNewWSClient asserted the old mid-stream reconfiguration; it now pins the
panic instead.
---
 transport/client_test.go |  4 +++-
 transport/connection.go  | 36 +++++++++++++++++++++++++++---------
 2 files changed, 30 insertions(+), 10 deletions(-)

diff --git a/transport/client_test.go b/transport/client_test.go
index 1cfe847a..80aa339e 100644
--- a/transport/client_test.go
+++ b/transport/client_test.go
@@ -503,7 +503,9 @@ func TestNewWSClient(t *testing.T) {
 
 	assert.Equal(t, 1, msgHandler.SessionNumber())
 	ss := msgHandler.array[0]
-	ss.SetCompressType(CompressNone)
+	// the session read loop already started IO, so reconfiguring compression
+	// mid-stream must be rejected, same contract as the TCP connection
+	assert.Panics(t, func() { ss.SetCompressType(CompressNone) })
 	conn := ss.(*session).Connection.(*gettyWSConn)
 	assert.True(t, conn.compress == CompressNone)
 	err := conn.handlePing("hello")
diff --git a/transport/connection.go b/transport/connection.go
index 37ec9c7c..2e0d28c8 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -86,8 +86,8 @@ var CodecStallTimeout = 5 * time.Minute
 type Connection interface {
 	ID() uint32
 	// SetCompressType sets the compress type. It must be called before any
-	// recv/Send on the connection, typically in NewSessionCallback; a TCP
-	// connection panics if it is called after IO started.
+	// recv/Send on the connection, typically in NewSessionCallback; TCP and
+	// websocket connections panic if it is called after IO started.
 	SetCompressType(CompressType)
 	LocalAddr() string
 	RemoteAddr() string
@@ -968,7 +968,12 @@ type gettyWSConn struct {
 	gettyConn
 	writeLock sync.Mutex
 	readLock  sync.Mutex
-	conn      *websocket.Conn
+	// streamStarted is set by the first read/write and freezes the compression
+	// configuration, mirroring the gettyTCPConn contract: gorilla's
+	// EnableWriteCompression/SetCompressionLevel are plain field writes that
+	// must not race with in-flight writers.
+	streamStarted uatomic.Bool
+	conn          *websocket.Conn
 }
 
 // create websocket connection
@@ -1003,18 +1008,29 @@ func newGettyWSConn(conn *websocket.Conn) *gettyWSConn {
 	return gettyWSConn
 }
 
-// SetCompressType set compress type
+// SetCompressType set compress type. Like the TCP variant it must be called
+// before the first recv/Send, typically in NewSessionCallback, and panics on a
+// started stream: gorilla's compression setters are plain field writes that
+// would race with in-flight writers.
 func (w *gettyWSConn) SetCompressType(c CompressType) {
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman:
-		w.conn.EnableWriteCompression(true)
-		if err := w.conn.SetCompressionLevel(int(c)); err != nil {
-			log.Warnf("failed to set compression level: %+v", err)
-		}
-
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
 	}
+	// writeLock excludes in-flight writers; the started check under it makes
+	// the panic race-free (writers mark streamStarted while holding the same
+	// lock, readers mark it under readLock before touching the conn).
+	w.writeLock.Lock()
+	defer w.writeLock.Unlock()
+	if w.streamStarted.Load() {
+		log.Errorf("SetCompressType(%d) called after IO started on connection{local:%s, peer:%s}", c, w.local, w.peer)
+		panic("SetCompressType must be called before any recv/Send on the connection, e.g. in NewSessionCallback")
+	}
+	w.conn.EnableWriteCompression(true)
+	if err := w.conn.SetCompressionLevel(int(c)); err != nil {
+		log.Warnf("failed to set compression level: %+v", err)
+	}
 	w.compress = c
 }
 
@@ -1135,6 +1151,7 @@ func (w *gettyWSConn) CloseConn(waitSec int) {
 func (w *gettyWSConn) threadSafeWriteMessage(messageType int, data []byte) error {
 	w.writeLock.Lock()
 	defer w.writeLock.Unlock()
+	w.streamStarted.Store(true)
 	if err := w.conn.WriteMessage(messageType, data); err != nil {
 		return err
 	}
@@ -1145,6 +1162,7 @@ func (w *gettyWSConn) threadSafeWriteMessage(messageType int, data []byte) error
 func (w *gettyWSConn) threadSafeReadMessage() (int, []byte, error) {
 	w.readLock.Lock()
 	defer w.readLock.Unlock()
+	w.streamStarted.Store(true)
 	messageType, readBytes, err := w.conn.ReadMessage()
 	if err != nil {
 		return messageType, nil, err

From 7da7a6cb436fffc8090c1008e4802b0bdc8f749a Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:04:23 +0800
Subject: [PATCH 10/15] fix: report 0 written bytes when a websocket Send fails

A failed WriteMessage delivers nothing, yet Send returned len(p) alongside
the error, so session.WritePkg counted the discarded frame as successCount.
---
 transport/connection.go         | 13 +++--
 transport/connection_ws_test.go | 86 +++++++++++++++++++++++++++++++++
 2 files changed, 94 insertions(+), 5 deletions(-)
 create mode 100644 transport/connection_ws_test.go

diff --git a/transport/connection.go b/transport/connection.go
index 2e0d28c8..6324a458 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -1107,11 +1107,14 @@ func (w *gettyWSConn) Send(pkg any) (int, error) {
 	if err := w.updateWriteDeadline(); err != nil {
 		log.Warnf("failed to update write deadline: %+v", err)
 	}
-	if err = w.threadSafeWriteMessage(websocket.BinaryMessage, p); err == nil {
-		w.writeBytes.Add((uint32)(len(p)))
-		w.writePkgNum.Add(1)
-	}
-	return len(p), perrors.WithStack(err)
+	// a failed WriteMessage delivers nothing (gorilla discards the frame), so
+	// report 0 written bytes: callers treat the count as the success count.
+	if err = w.threadSafeWriteMessage(websocket.BinaryMessage, p); err != nil {
+		return 0, perrors.WithStack(err)
+	}
+	w.writeBytes.Add((uint32)(len(p)))
+	w.writePkgNum.Add(1)
+	return len(p), nil
 }
 
 func (w *gettyWSConn) writePing() error {
diff --git a/transport/connection_ws_test.go b/transport/connection_ws_test.go
new file mode 100644
index 00000000..e129dc68
--- /dev/null
+++ b/transport/connection_ws_test.go
@@ -0,0 +1,86 @@
+/*
+ * 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 (
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+)
+
+import (
+	"github.com/gorilla/websocket"
+)
+
+func newWSClientConn(t *testing.T) *gettyWSConn {
+	t.Helper()
+
+	upgrader := websocket.Upgrader{}
+	serverConnCh := make(chan *websocket.Conn, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		conn, err := upgrader.Upgrade(w, r, nil)
+		if err != nil {
+			return
+		}
+		serverConnCh <- conn
+	}))
+	t.Cleanup(srv.Close)
+
+	clientWS, resp, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http"), nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if resp != nil && resp.Body != nil {
+		_ = resp.Body.Close()
+	}
+	serverWS := <-serverConnCh
+	t.Cleanup(func() {
+		_ = clientWS.Close()
+		_ = serverWS.Close()
+	})
+	return newGettyWSConn(clientWS)
+}
+
+// TestWSSendReturnsZeroOnError pins the write count contract: a failed
+// WriteMessage delivers nothing, so Send must report 0 written bytes -
+// returning len(p) alongside the error inflates the caller's success count
+// (session.WritePkg uses it as successCount).
+func TestWSSendReturnsZeroOnError(t *testing.T) {
+	client := newWSClientConn(t)
+
+	payload := []byte("ws-payload")
+	if n, err := client.Send(payload); err != nil || n != len(payload) {
+		t.Fatalf("healthy Send = (%d, %v), want (%d, nil)", n, err, len(payload))
+	}
+	if got := client.writePkgNum.Load(); got != 1 {
+		t.Fatalf("writePkgNum = %d after one successful Send, want 1", got)
+	}
+
+	_ = client.conn.UnderlyingConn().Close()
+	n, err := client.Send(payload)
+	if err == nil {
+		t.Fatal("Send on a closed websocket returned no error")
+	}
+	if n != 0 {
+		t.Fatalf("failed Send reported %d written bytes, want 0", n)
+	}
+	if got := client.writePkgNum.Load(); got != 1 {
+		t.Fatalf("failed Send bumped writePkgNum to %d, want it unchanged at 1", got)
+	}
+}

From 9a242d8b4ff7f2e98f88a8fa819591fc4b2fdd33 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:08:54 +0800
Subject: [PATCH 11/15] fix: stop pretending UDP connections support
 compression

UDP recv/Send never ran a codec, but SetCompressType accepted every type
silently, so callers believed compression was on. Warn instead, and record
the type in the embedded gettyConn.compress like TCP/WS do: the shadow
compressType field was written and never read, leaving compress stuck at
CompressNone.
---
 transport/connection.go      | 16 ++++++++++++----
 transport/connection_test.go | 17 +++++++++++++++++
 2 files changed, 29 insertions(+), 4 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 6324a458..e46afe10 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -844,9 +844,8 @@ func (c UDPContext) String() string {
 
 type gettyUDPConn struct {
 	gettyConn
-	compressType CompressType
-	conn         *net.UDPConn // for server; immutable after construction, closed via closeOnce
-	closeOnce    sync.Once
+	conn      *net.UDPConn // for server; immutable after construction, closed via closeOnce
+	closeOnce sync.Once
 }
 
 // create gettyUDPConn
@@ -878,10 +877,19 @@ func newGettyUDPConn(conn *net.UDPConn) *gettyUDPConn {
 	}
 }
 
+// SetCompressType records the requested type but UDP send/recv never
+// compress: each datagram is an independent packet with no stream to run a
+// codec over, and this implementation never had one. Accepting the call
+// silently made callers believe compression was on, so any type other than
+// CompressNone is now loudly reported as unsupported (still accepted, not a
+// panic, to keep existing callers running).
 func (u *gettyUDPConn) SetCompressType(c CompressType) {
 	switch c {
 	case CompressNone, CompressZip, CompressBestSpeed, CompressBestCompression, CompressHuffman, CompressSnappy:
-		u.compressType = c
+		if c != CompressNone {
+			log.Warnf("UDP connection{local:%s, peer:%s} does not support compression, SetCompressType(%d) has no effect on the wire", u.local, u.peer, c)
+		}
+		u.compress = c
 
 	default:
 		panic(fmt.Sprintf("illegal comparess type %d", c))
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 009f24d8..ed0a03e4 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -895,3 +895,20 @@ func TestSetCompressTypeConcurrentWithSend(t *testing.T) {
 		t.Fatal("timed out waiting for the stream")
 	}
 }
+
+// TestUDPSetCompressTypeHasNoWireEffect pins the UDP contract: compression is
+// not supported, the call records the type (and warns) instead of silently
+// pretending or panicking under existing callers.
+func TestUDPSetCompressTypeHasNoWireEffect(t *testing.T) {
+	raw, err := net.ListenUDP("udp", &net.UDPAddr{IP: net.IPv4(127, 0, 0, 1)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	conn := newGettyUDPConn(raw)
+	t.Cleanup(func() { conn.CloseConn(0) })
+
+	conn.SetCompressType(CompressSnappy)
+	if conn.compress != CompressSnappy {
+		t.Fatalf("compress = %d, want %d recorded", conn.compress, CompressSnappy)
+	}
+}

From 20af863a9ced6a019779c802035562f1385b9b29 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:11:58 +0800
Subject: [PATCH 12/15] fix: bound the codec flush in CloseConn instead of
 waiting out a stall window

CloseConn flushes the codec writer through codecPollingWriter, which absorbs
poll timeouts for up to codecStallTimeout (5 min by default). Against a peer
that stopped reading, a direct CloseConn - no session to report IsClosed() -
was pinned for that whole window. Mark the conn as closing so the flush gives
up on the first poll timeout.
---
 transport/connection.go      | 12 +++++++++++-
 transport/connection_test.go | 31 +++++++++++++++++++++++++++++++
 2 files changed, 42 insertions(+), 1 deletion(-)

diff --git a/transport/connection.go b/transport/connection.go
index e46afe10..d2b467f2 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -233,6 +233,12 @@ type gettyTCPConn struct {
 	// it has to be atomic. Once set, recv/Send fail fast: touching the codec
 	// again would only produce more garbage on the wire.
 	codecBroken uatomic.Bool
+	// closing is set by CloseConn before it flushes the codec writer: the
+	// flush is best-effort, so codecPollingWriter surfaces the first poll
+	// timeout instead of waiting out a whole codecStallTimeout on a peer that
+	// stopped reading (sessionClosing() covers the session-driven close, this
+	// covers direct CloseConn use).
+	closing uatomic.Bool
 	// codecStallTimeout is the per-conn copy of CodecStallTimeout, taken when
 	// the codec is installed. 0 disables mid-block stall detection.
 	codecStallTimeout time.Duration
@@ -583,7 +589,7 @@ func (w *codecPollingWriter) Write(p []byte) (int, error) {
 		if !isTimeoutError(err) {
 			return written, err
 		}
-		if t.codecBroken.Load() || t.sessionClosing() {
+		if t.codecBroken.Load() || t.closing.Load() || t.sessionClosing() {
 			// on shutdown the write is abandoned mid-block; the connection is
 			// being torn down anyway, and latching codecBroken here also stops
 			// CloseConn from flushing the half-written snappy stream.
@@ -802,6 +808,10 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 // close tcp connection
 func (t *gettyTCPConn) CloseConn(waitSec int) {
 	t.closeOnce.Do(func() {
+		// best-effort teardown from here on: codecPollingWriter surfaces the
+		// first poll timeout instead of waiting out a stall window, so a peer
+		// that stopped reading cannot pin CloseConn for codecStallTimeout.
+		t.closing.Store(true)
 		t.lock.Lock()
 		writer := t.writer
 		t.lock.Unlock()
diff --git a/transport/connection_test.go b/transport/connection_test.go
index ed0a03e4..2a2d9c13 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -912,3 +912,34 @@ func TestUDPSetCompressTypeHasNoWireEffect(t *testing.T) {
 		t.Fatalf("compress = %d, want %d recorded", conn.compress, CompressSnappy)
 	}
 }
+
+// TestCloseConnCodecFlushIsBounded pins the closing flag: CloseConn flushes the
+// codec writer best-effort, so a peer that stopped reading must cost at most
+// one poll timeout, never a whole codecStallTimeout.
+func TestCloseConnCodecFlushIsBounded(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(20 * time.Millisecond)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 5 * time.Second
+
+			done := make(chan struct{})
+			go func() {
+				client.CloseConn(0)
+				close(done)
+			}()
+			select {
+			case <-done:
+			case <-time.After(2 * time.Second):
+				t.Fatal("CloseConn blocked on the codec flush instead of giving up after one poll timeout")
+			}
+		})
+	}
+}

From 1a72e70e6135d6ddf0deb4300668ecbbef64f0f0 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:12:39 +0800
Subject: [PATCH 13/15] fix: terminate the flate stream on CloseConn like the
 snappy one

CloseConn closed only the snappy writer, so a flate peer hit
io.ErrUnexpectedEOF at the end of an otherwise clean shutdown: no data was
lost (every Write flushes) but the stream never got its final block marker.
---
 transport/connection.go      | 21 ++++++++++++++++++---
 transport/connection_test.go | 26 ++++++++++++++++++++++++++
 2 files changed, 44 insertions(+), 3 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index d2b467f2..b02da38a 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -326,6 +326,15 @@ func (t *writeFlusher) WriteBuffers(buffers [][]byte) (int64, error) {
 	return total, perrors.WithStack(t.flusher.Flush())
 }
 
+// Close terminates the flate stream (final block marker), so the peer's
+// decoder sees a clean EOF instead of io.ErrUnexpectedEOF. Mirrors
+// snappyWriteFlusher.Close; called by CloseConn on a healthy codec only.
+func (t *writeFlusher) Close() error {
+	t.lock.Lock()
+	defer t.lock.Unlock()
+	return perrors.WithStack(t.flusher.Close())
+}
+
 // 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
@@ -815,15 +824,21 @@ func (t *gettyTCPConn) CloseConn(waitSec int) {
 		t.lock.Lock()
 		writer := t.writer
 		t.lock.Unlock()
-		// #102: snappy writer is now wrapped in *snappyWriteFlusher.
+		// #102: the codec writers are wrapped in flushers with a Close that
+		// terminates the stream, giving the peer's decoder a clean EOF.
 		// A broken codec must not be flushed: the stream is already
 		// desynchronized, and Close would only push more garbage into a socket
 		// that may still be stalled.
 		if !t.codecBroken.Load() {
-			if writer, ok := writer.(*snappyWriteFlusher); ok {
-				if err := writer.Close(); err != nil {
+			switch w := writer.(type) {
+			case *snappyWriteFlusher:
+				if err := w.Close(); err != nil {
 					log.Errorf("snappy.Writer.Close() = error:%+v", err)
 				}
+			case *writeFlusher:
+				if err := w.Close(); err != nil {
+					log.Errorf("flate.Writer.Close() = error:%+v", err)
+				}
 			}
 		}
 		// #103: do not hard-assert *tls.Conn; use safe type assertions so a
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 2a2d9c13..7ed3021b 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -943,3 +943,29 @@ func TestCloseConnCodecFlushIsBounded(t *testing.T) {
 		})
 	}
 }
+
+// TestCloseConnTerminatesFlateStreamCleanly pins the write-side symmetry with
+// snappy: CloseConn closes the flate writer too, so the peer's decoder reads
+// the payload and then a clean EOF instead of io.ErrUnexpectedEOF.
+func TestCloseConnTerminatesFlateStreamCleanly(t *testing.T) {
+	client, server := newTCPConnPair(t)
+	client.SetWriteTimeout(time.Second)
+	client.SetCompressType(CompressZip)
+
+	payload := []byte("last-packet-before-close")
+	if _, err := client.Send(payload); err != nil {
+		t.Fatalf("Send failed: %v", err)
+	}
+	client.CloseConn(1)
+
+	if err := server.conn.SetReadDeadline(time.Now().Add(5 * time.Second)); err != nil {
+		t.Fatal(err)
+	}
+	got, err := io.ReadAll(flate.NewReader(server.conn))
+	if err != nil {
+		t.Fatalf("peer decoder did not see a clean end of stream: %v", err)
+	}
+	if !bytes.Equal(got, payload) {
+		t.Fatalf("decoded %q, want %q", got, payload)
+	}
+}

From 330951f217598a945679292b3c5546dd47010b11 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:13:21 +0800
Subject: [PATCH 14/15] fix: clamp the codec poll interval to the stall
 deadline

Stall detection only runs when a poll wakes up, so an rTimeout/wTimeout larger
than codecStallTimeout pushed detection out by a whole poll interval - a 30s
read timeout meant a mid-block stall sat undetected for 30s regardless of the
5 min bound. Cap the deadline at the remaining stall window.
---
 transport/connection.go      | 16 ++++++++
 transport/connection_test.go | 73 ++++++++++++++++++++++++++++++++++++
 2 files changed, 89 insertions(+)

diff --git a/transport/connection.go b/transport/connection.go
index b02da38a..3ae9d660 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -519,6 +519,14 @@ func (r *codecPollingReader) fill() error {
 	t := r.t
 	for {
 		if timeout := t.rTimeout.Load(); timeout > 0 {
+			// mid-block the poll must wake up no later than the stall deadline,
+			// otherwise an rTimeout larger than codecStallTimeout would delay
+			// stall detection by up to a whole poll interval.
+			if stall := t.codecStallTimeout; r.progress && stall > 0 {
+				if remaining := stall - time.Since(r.lastDelivery); remaining < timeout {
+					timeout = max(remaining, time.Millisecond)
+				}
+			}
 			// Set Deadline every time, since golang has fixed the performance issue
 			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 			currentTime := time.Now()
@@ -578,6 +586,14 @@ func (w *codecPollingWriter) Write(p []byte) (int, error) {
 	lastProgress := time.Now()
 	for written < len(p) {
 		if timeout := t.wTimeout.Load(); timeout > 0 {
+			// the poll must wake up no later than the stall deadline, otherwise
+			// a wTimeout larger than codecStallTimeout would delay stall
+			// detection by up to a whole poll interval.
+			if stall := t.codecStallTimeout; stall > 0 {
+				if remaining := stall - time.Since(lastProgress); remaining < timeout {
+					timeout = max(remaining, time.Millisecond)
+				}
+			}
 			// Set Deadline every time, since golang has fixed the performance issue
 			// See https://github.com/golang/go/issues/15133#issuecomment-271571395 for details
 			currentTime := time.Now()
diff --git a/transport/connection_test.go b/transport/connection_test.go
index 7ed3021b..5dc56820 100644
--- a/transport/connection_test.go
+++ b/transport/connection_test.go
@@ -969,3 +969,76 @@ func TestCloseConnTerminatesFlateStreamCleanly(t *testing.T) {
 		t.Fatalf("decoded %q, want %q", got, payload)
 	}
 }
+
+// TestCodecRecvStallDetectionOutrunsLongPollInterval pins the poll clamp: when
+// the stream is mid-block, the poll deadline is capped at the remaining stall
+// window, so an rTimeout larger than codecStallTimeout (here 10s vs 200ms) must
+// not delay stall detection until the next full poll interval.
+func TestCodecRecvStallDetectionOutrunsLongPollInterval(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			client, server := newTCPConnPair(t)
+			client.SetReadTimeout(10 * time.Second)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			if _, err := server.conn.Write(halfCodecBlock(t, test.compress, "stalled-peer-payload")); err != nil {
+				t.Fatalf("peer write failed: %v", err)
+			}
+
+			recvErr := make(chan error, 1)
+			go func() {
+				buf := make([]byte, 64)
+				for callsLeft := 1000; callsLeft > 0; callsLeft-- {
+					if _, err := client.recv(buf); err != nil {
+						recvErr <- err
+						return
+					}
+				}
+				recvErr <- nil
+			}()
+
+			var err error
+			select {
+			case err = <-recvErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("stall not detected within 3s: the poll interval was not clamped to the stall deadline")
+			}
+			assertCodecStreamBroken(t, err)
+		})
+	}
+}
+
+// TestCodecSendStallDetectionOutrunsLongPollInterval is the write-side twin of
+// the poll clamp test: a peer that never reads must be detected within the
+// stall window even when wTimeout is far larger than codecStallTimeout.
+func TestCodecSendStallDetectionOutrunsLongPollInterval(t *testing.T) {
+	for _, test := range codecStallCases {
+		t.Run(test.name, func(t *testing.T) {
+			clientRaw, peerRaw := net.Pipe()
+			t.Cleanup(func() {
+				_ = clientRaw.Close()
+				_ = peerRaw.Close()
+			})
+
+			client := newGettyTCPConn(clientRaw)
+			client.SetWriteTimeout(10 * time.Second)
+			client.SetCompressType(test.compress)
+			client.codecStallTimeout = 200 * time.Millisecond
+
+			sendErr := make(chan error, 1)
+			go func() {
+				_, err := client.Send([]byte("peer never reads this"))
+				sendErr <- err
+			}()
+
+			var err error
+			select {
+			case err = <-sendErr:
+			case <-time.After(3 * time.Second):
+				t.Fatal("stall not detected within 3s: the poll interval was not clamped to the stall deadline")
+			}
+			assertCodecStreamBroken(t, err)
+		})
+	}
+}

From 887e13d989ab0b4388b2a1babb0e39a00bf61337 Mon Sep 17 00:00:00 2001
From: bang <3656828039@qq.com>
Date: Thu, 20 Aug 2026 03:18:11 +0800
Subject: [PATCH 15/15] docs: correct pkg counter comments, drop zero-value
 time from codec Send logs

readPkgNum/writePkgNum had their comments swapped. The Send debug logs printed
currentTime, which stays zero on a codec stream (only a raw conn arms the
deadline there) - the log framework stamps the time anyway. Also record the one
remaining blind spot in the mid-block stall detection, so the guarantee reads
as what it is.
---
 transport/connection.go | 22 ++++++++++++++++------
 1 file changed, 16 insertions(+), 6 deletions(-)

diff --git a/transport/connection.go b/transport/connection.go
index 3ae9d660..2d161d35 100644
--- a/transport/connection.go
+++ b/transport/connection.go
@@ -130,8 +130,8 @@ type gettyConn struct {
 	codecEnabled  bool
 	readBytes     uatomic.Uint32   // read bytes
 	writeBytes    uatomic.Uint32   // write bytes
-	readPkgNum    uatomic.Uint32   // send pkg number
-	writePkgNum   uatomic.Uint32   // recv pkg number
+	readPkgNum    uatomic.Uint32   // recv pkg number
+	writePkgNum   uatomic.Uint32   // send pkg number
 	active        uatomic.Int64    // last active, in milliseconds
 	rTimeout      uatomic.Duration // network current limiting
 	wTimeout      uatomic.Duration
@@ -479,6 +479,16 @@ var _ flate.Reader = (*codecPollingReader)(nil)
 // boundary is called by recv after the codec returned decoded output: the
 // stream is at a block boundary again, so subsequent silence is idleness, not
 // a stall.
+//
+// "At a block boundary" is exact for flushed packets: flate only surfaces the
+// decoded output after it consumed the sync-flush empty stored block, and
+// snappy only after a whole chunk. The one approximation is flate's 32KB
+// window flush, which surfaces output mid-block; that is still covered
+// because the block's remaining bytes sit undelivered in r.buf and re-mark
+// progress on the next Read - EXCEPT when the peer died byte-exactly at such
+// a flush point with nothing left in r.buf. That coincidence degrades to a
+// hang indistinguishable from a peer dying between packets, which no
+// conn-level detection can catch; application heartbeats own that case.
 func (r *codecPollingReader) boundary() {
 	r.progress = false
 }
@@ -811,8 +821,8 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 			t.writeBytes.Add((uint32)(lg))
 			t.writePkgNum.Add((uint32)(len(buffers)))
 		}
-		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%v",
-			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, lg, err)
+		log.Debugf("localAddr: %s, remoteAddr:%s, length:%d, err:%v",
+			t.conn.LocalAddr(), t.conn.RemoteAddr(), lg, err)
 		return int(lg), t.codecIOError(err)
 	}
 
@@ -822,8 +832,8 @@ func (t *gettyTCPConn) Send(pkg any) (int, error) {
 			t.writeBytes.Add((uint32)(len(p)))
 			t.writePkgNum.Add(1)
 		}
-		log.Debugf("localAddr: %s, remoteAddr:%s, now:%s, length:%d, err:%v",
-			t.conn.LocalAddr(), t.conn.RemoteAddr(), currentTime, length, err)
+		log.Debugf("localAddr: %s, remoteAddr:%s, length:%d, err:%v",
+			t.conn.LocalAddr(), t.conn.RemoteAddr(), length, err)
 		return length, t.codecIOError(err)
 	}