From 967a7505aef0af02d0392b65a74e392fcd4406e2 Mon Sep 17 00:00:00 2001 From: fatedier Date: Tue, 3 Nov 2026 21:28:54 +0800 Subject: [PATCH] deps: bump fatedier/yamux to v0.3.0 (#5595) Update github.com/fatedier/yamux from v0.2.0 to v0.3.0. go.mod and go.sum are the only dependency changes. v0.3.0 adds a public Stream.Reset() and fixes several stream and session lifecycle problems: a read or write blocked on a shutting-down session returns instead of spinning, the stream close lifecycle is safe to retry, failed stream opens release their tracking state, and ConnectionWriteTimeout is now enforced as a write deadline on the underlying transport instead of only bounding how long a caller waits for its own frame. The last change is observable in frp. When a tcpMux connection stalls, the write now fails at the transport and the session is torn down, so a new stream returns ErrSessionShutdown; before, the caller timed out while the session stayed blocked but alive until a keepalive ping also timed out. All frp transports implement SetWriteDeadline (TCP, TLS, WebSocket, WSS and KCP); KCP reports its own timeout error instead of the mapped one. Reads on an established stream now return ErrSessionShutdown instead of io.EOF when the session dies; a graceful stream close still returns io.EOF, and frp data paths already treat any read error as the end of the stream. Adapt TestDispatcherYamuxWriteTimeoutInterruptsRead to the new behaviour: the delayed test connection now honours SetWriteDeadline, and the test asserts that both yamux sessions close after the timed-out write instead of expecting the session to stay usable. --- go.mod | 2 +- go.sum | 4 +-- pkg/msg/handler_yamux_test.go | 52 ++++++++++++++++++++++++----------- 3 files changed, 39 insertions(+), 19 deletions(-) diff --git a/go.mod b/go.mod index 228b3b21..4920212f 100644 --- a/go.mod +++ b/go.mod @@ -6,7 +6,7 @@ require ( github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5 github.com/coreos/go-oidc/v3 v3.18.0 github.com/fatedier/golib v0.8.2 - github.com/fatedier/yamux v0.2.0 + github.com/fatedier/yamux v0.3.0 github.com/google/uuid v1.6.0 github.com/gorilla/mux v1.8.1 github.com/gorilla/websocket v1.5.0 diff --git a/go.sum b/go.sum index e5b5ba7e..80da683e 100644 --- a/go.sum +++ b/go.sum @@ -22,8 +22,8 @@ github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1m github.com/envoyproxy/protoc-gen-validate v0.1.0/go.mod h1:iSmxcyjqTsJpI2R4NaDN7+kN2VEUnK/pcBlmesArF7c= github.com/fatedier/golib v0.8.2 h1:02n2Dg7KJ7rR7p7n4/6hBUjaLQf2J7EiHYZQsgGTvww= github.com/fatedier/golib v0.8.2/go.mod h1:ArUGvPg2cOw/py2RAuBt46nNZH2VQ5Z70p109MAZpJw= -github.com/fatedier/yamux v0.2.0 h1:H+2A9iBVh7aJlEOc1Ws1FXWOaecBf2nRv9zpFMPUWg8= -github.com/fatedier/yamux v0.2.0/go.mod h1:d4FtRDrC9sHvRpiDL6J5EnfjLhqzZZplHe5yToSn2Ac= +github.com/fatedier/yamux v0.3.0 h1:NHcAbiv2YMZJjV413Ibw0OccvRVWP1nTTQjdRdFKI/0= +github.com/fatedier/yamux v0.3.0/go.mod h1:d4FtRDrC9sHvRpiDL6J5EnfjLhqzZZplHe5yToSn2Ac= github.com/go-jose/go-jose/v4 v4.1.4 h1:moDMcTHmvE6Groj34emNPLs/qtYXRVcd6S7NHbHz3kA= github.com/go-jose/go-jose/v4 v4.1.4/go.mod h1:x4oUasVrzR7071A4TnHLGSPpNOm2a21K9Kf04k1rs08= github.com/go-logr/logr v1.4.2 h1:6pFjapn8bFcIbiKo3XT4j/BhANplGihG6tvd+8rYgrY= diff --git a/pkg/msg/handler_yamux_test.go b/pkg/msg/handler_yamux_test.go index 25d91a9e..98e0a850 100644 --- a/pkg/msg/handler_yamux_test.go +++ b/pkg/msg/handler_yamux_test.go @@ -18,6 +18,7 @@ import ( "errors" "io" "net" + "os" "sync" "sync/atomic" "testing" @@ -30,18 +31,45 @@ import ( ) // Delay one real TCP write without injecting an error. The stream's write -// failure must come from yamux's ConnectionWriteTimeout. +// failure must come from yamux's ConnectionWriteTimeout, which yamux enforces +// as a write deadline on the underlying connection, so the delayed write has +// to observe that deadline the way a real socket does. type delayedTCPConn struct { net.Conn armed atomic.Bool entered chan struct{} release chan struct{} + + deadlineMu sync.Mutex + deadline time.Time +} + +func (c *delayedTCPConn) SetWriteDeadline(t time.Time) error { + c.deadlineMu.Lock() + c.deadline = t + c.deadlineMu.Unlock() + return c.Conn.SetWriteDeadline(t) } func (c *delayedTCPConn) Write(p []byte) (int, error) { if c.armed.Swap(false) { close(c.entered) - <-c.release + c.deadlineMu.Lock() + deadline := c.deadline + c.deadlineMu.Unlock() + var timeout <-chan time.Time + if !deadline.IsZero() { + timer := time.NewTimer(time.Until(deadline)) + defer timer.Stop() + timeout = timer.C + } + select { + case <-c.release: + case <-timeout: + // Same value a real socket returns; yamux maps any timeout error + // to ErrConnectionWriteTimeout. + return 0, os.ErrDeadlineExceeded + } } return c.Conn.Write(p) } @@ -206,20 +234,12 @@ func TestDispatcherYamuxWriteTimeoutInterruptsRead(t *testing.T) { if observed.closeCalls.Load() != 1 { t.Fatalf("control closed %d times, want once", observed.closeCalls.Load()) } - if client.IsClosed() || server.IsClosed() { - t.Fatal("yamux session unexpectedly closed") - } - fresh, freshPeer, freshObserved := loopbackControlPair(t, client, server, protocol) - for range 5 { - if err := fresh.WriteMsg(&ReqWorkConn{}); err != nil { - t.Fatal(err) - } - loopbackReadReq(t, freshPeer) - for len(freshObserved.writes) > 0 { - <-freshObserved.writes - } - } - t.Log("Done/readLoop exited without peer FIN; Close once; same yamux session decrypted five fresh messages") + // yamux enforces ConnectionWriteTimeout on the transport, so the + // failed write ends the whole session instead of leaving it blocked + // but alive. + regressionAwait(t, client.CloseChan(), "client yamux session close") + regressionAwait(t, server.CloseChan(), "server yamux session close") + t.Log("Done/readLoop exited without peer FIN; Close once; write timeout closed both yamux sessions") }) } }