mirror of
https://github.com/go-gost/x.git
synced 2026-10-19 16:16:32 +00:00
mux: a stream's deadline must not reach the shared session conn
streamConn embeds the session's net.Conn and overrides Read/Write/Close, but not the deadline methods, so those reached the connection every stream shares. Any deadline set by one stream then outlived it and decided when the others could read. net/http makes this reachable: tearing down a connection parks a pending read with SetReadDeadline(time.Unix(1, 0)) and clears it only when a read was actually in flight. Without a read in flight the shared conn is left with a deadline in 1970, and the next Accept on the session fails immediately with an i/o timeout — reported as a listener error even though the session was healthy. Delegate the three deadline methods to the stream, which has always had its own. The test drives one request end to end over a stream, applies that teardown deadline, and asserts the session still hands out the next stream. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Opus 4.8
parent
2306f45bf9
commit
8683da6a65
@@ -188,6 +188,28 @@ func (c *streamConn) Close() error {
|
||||
return c.stream.Close()
|
||||
}
|
||||
|
||||
// SetReadDeadline sets the deadline on this stream, never on the session conn
|
||||
// underneath. The embedded net.Conn is shared by every stream, so a deadline
|
||||
// set there outlives the stream that set it: net/http parks a read with
|
||||
// SetReadDeadline(time.Unix(1, 0)) to unblock it on teardown and only clears
|
||||
// it when a read was actually in flight, leaving the shared conn with a
|
||||
// deadline in 1970 — and the next session Accept fails with an i/o timeout.
|
||||
func (c *streamConn) SetReadDeadline(t time.Time) error {
|
||||
return c.stream.SetReadDeadline(t)
|
||||
}
|
||||
|
||||
// SetWriteDeadline sets the deadline on this stream, never on the session conn
|
||||
// underneath. See SetReadDeadline.
|
||||
func (c *streamConn) SetWriteDeadline(t time.Time) error {
|
||||
return c.stream.SetWriteDeadline(t)
|
||||
}
|
||||
|
||||
// SetDeadline sets both deadlines on this stream, never on the session conn
|
||||
// underneath. See SetReadDeadline.
|
||||
func (c *streamConn) SetDeadline(t time.Time) error {
|
||||
return c.stream.SetDeadline(t)
|
||||
}
|
||||
|
||||
func (c *streamConn) Context() context.Context {
|
||||
if sc, ok := c.Conn.(ctx.Context); ok {
|
||||
return sc.Context()
|
||||
@@ -291,6 +313,25 @@ func (c *yamuxStreamConn) Close() error {
|
||||
return c.stream.Close()
|
||||
}
|
||||
|
||||
// SetReadDeadline sets the deadline on this stream, never on the session conn
|
||||
// underneath. See streamConn.SetReadDeadline for why a leaked deadline here
|
||||
// breaks every other stream on the session.
|
||||
func (c *yamuxStreamConn) SetReadDeadline(t time.Time) error {
|
||||
return c.stream.SetReadDeadline(t)
|
||||
}
|
||||
|
||||
// SetWriteDeadline sets the deadline on this stream, never on the session conn
|
||||
// underneath. See streamConn.SetReadDeadline.
|
||||
func (c *yamuxStreamConn) SetWriteDeadline(t time.Time) error {
|
||||
return c.stream.SetWriteDeadline(t)
|
||||
}
|
||||
|
||||
// SetDeadline sets both deadlines on this stream, never on the session conn
|
||||
// underneath. See streamConn.SetReadDeadline.
|
||||
func (c *yamuxStreamConn) SetDeadline(t time.Time) error {
|
||||
return c.stream.SetDeadline(t)
|
||||
}
|
||||
|
||||
func (c *yamuxStreamConn) Context() context.Context {
|
||||
if sc, ok := c.Conn.(ctx.Context); ok {
|
||||
return sc.Context()
|
||||
|
||||
@@ -0,0 +1,104 @@
|
||||
package mux
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// A per-stream deadline must never reach the session conn underneath: the
|
||||
// session is shared by every stream, so one stream's deadline would decide when
|
||||
// the others may read. net/http makes this concrete — it parks a pending read
|
||||
// with SetReadDeadline(time.Unix(1, 0)) to unblock it on teardown, and only
|
||||
// clears that deadline when a read was actually in flight. A stream conn that
|
||||
// passes the call through therefore leaves the session deadline in 1970, and
|
||||
// the next Accept fails with an i/o timeout.
|
||||
func TestStreamDeadlineStaysOnTheStream(t *testing.T) {
|
||||
for _, backend := range []struct {
|
||||
name string
|
||||
client func(net.Conn) (Session, error)
|
||||
server func(net.Conn) (Session, error)
|
||||
}{
|
||||
{"smux", func(c net.Conn) (Session, error) { return ClientSession(c, &Config{Version: 2}) },
|
||||
func(c net.Conn) (Session, error) { return ServerSession(c, &Config{Version: 2}) }},
|
||||
{"yamux", func(c net.Conn) (Session, error) { return ClientSession(c, &Config{Type: "yamux"}) },
|
||||
func(c net.Conn) (Session, error) { return ServerSession(c, &Config{Type: "yamux"}) }},
|
||||
} {
|
||||
t.Run(backend.name, func(t *testing.T) {
|
||||
cli, srv := net.Pipe()
|
||||
defer cli.Close()
|
||||
|
||||
srvSession, err := backend.server(srv)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer srvSession.Close()
|
||||
|
||||
cliSession, err := backend.client(cli)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
// One request served end to end over a stream, then the stream is
|
||||
// torn down the way net/http tears one down.
|
||||
stream, err := cliSession.GetConn()
|
||||
if err != nil {
|
||||
t.Fatal("open stream: ", err)
|
||||
}
|
||||
|
||||
srvConn, err := srvSession.Accept()
|
||||
if err != nil {
|
||||
t.Fatal("accept stream: ", err)
|
||||
}
|
||||
go func() {
|
||||
io.WriteString(srvConn, "HTTP/1.0 200 OK\r\nContent-Length: 2\r\n\r\nok")
|
||||
srvConn.Close()
|
||||
}()
|
||||
|
||||
req, err := http.NewRequestWithContext(context.Background(), "GET", "http://x/", nil)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
resp, err := (&http.Client{
|
||||
Transport: &http.Transport{
|
||||
DialContext: func(context.Context, string, string) (net.Conn, error) { return stream, nil },
|
||||
},
|
||||
}).Do(req)
|
||||
if err != nil {
|
||||
t.Fatal("request: ", err)
|
||||
}
|
||||
body, err := io.ReadAll(resp.Body)
|
||||
resp.Body.Close()
|
||||
if err != nil {
|
||||
t.Fatal("read body: ", err)
|
||||
}
|
||||
if string(body) != "ok" {
|
||||
t.Fatalf("body = %q, want %q", body, "ok")
|
||||
}
|
||||
|
||||
// net/http's teardown deadline: unblock a pending read by parking
|
||||
// it in the past, then clear it only if a read was in flight.
|
||||
stream.SetReadDeadline(time.Unix(1, 0))
|
||||
stream.SetReadDeadline(time.Time{})
|
||||
|
||||
// The session must still hand out the next stream: its deadline was
|
||||
// never touched by the one that just closed.
|
||||
done := make(chan error, 1)
|
||||
go func() {
|
||||
_, err := cliSession.GetConn()
|
||||
done <- err
|
||||
}()
|
||||
select {
|
||||
case err := <-done:
|
||||
if err != nil {
|
||||
t.Fatalf("session unusable after a served stream: %v", err)
|
||||
}
|
||||
case <-time.After(5 * time.Second):
|
||||
t.Fatal("session hung after a served stream")
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user