From 815c9c57726d0c5df77878da3f49a5d6b95c6c46 Mon Sep 17 00:00:00 2001 From: MHSanaei Date: Sat, 3 Oct 2026 15:48:05 +0200 Subject: [PATCH] fix(tuic): accept the server's STOP_SENDING when tests close uni streams The race job failed in TestAudit3ManagerEnsureActualSendersWithPersistentTraffic with "close called for canceled stream 14". The server parses one command per uni stream and then calls CancelRead, as the quinn reference server does on drop, so its STOP_SENDING can reach the client before the client's own Close and quic-go reports that Close as an error. The data was already read. Every test that wrote a command on a uni stream and required Close to succeed shared this race. closeUniStream accepts only a remote StreamError on the stream's context, so any other Close failure still fails the test. --- internal/tuic/manager_live_traffic_test.go | 8 ++---- internal/tuic/relay_logging_test.go | 4 +-- internal/tuic/server_test.go | 31 +++++++++++++++++----- internal/tuic/stream_udp_lifecycle_test.go | 4 +-- 4 files changed, 29 insertions(+), 18 deletions(-) diff --git a/internal/tuic/manager_live_traffic_test.go b/internal/tuic/manager_live_traffic_test.go index 146c208d3..5403ca856 100644 --- a/internal/tuic/manager_live_traffic_test.go +++ b/internal/tuic/manager_live_traffic_test.go @@ -96,9 +96,7 @@ func TestAudit3ManagerEnsureActualSendersWithPersistentTraffic(t *testing.T) { if _, err := stream.Write(frame.Bytes()); err != nil { t.Fatal(err) } - if err := stream.Close(); err != nil { - t.Fatal(err) - } + closeUniStream(t, stream) response, err := p.client.AcceptUniStream(ctx) if err != nil { t.Fatal(err) @@ -164,9 +162,7 @@ func TestAudit3ManagerEnsureActualSendersWithPersistentTraffic(t *testing.T) { if _, err := auth.Write(authBytes); err != nil { t.Fatal(err) } - if err := auth.Close(); err != nil { - t.Fatal(err) - } + closeUniStream(t, auth) waitForClientCongestionSender(t, server, client, served) var serverConn *quic.Conn diff --git a/internal/tuic/relay_logging_test.go b/internal/tuic/relay_logging_test.go index 9493a6cf8..9c41f44f7 100644 --- a/internal/tuic/relay_logging_test.go +++ b/internal/tuic/relay_logging_test.go @@ -57,9 +57,7 @@ func audit3LogsStart(t *testing.T, level, marker, relayAddr string) (*Server, *c if _, err := auth.Write(frame); err != nil { t.Fatal(err) } - if err := auth.Close(); err != nil { - t.Fatal(err) - } + closeUniStream(t, auth) _, _ = authenticatedServerConnection(t, s, id) return s, c, id, password, token } diff --git a/internal/tuic/server_test.go b/internal/tuic/server_test.go index baf505d79..774d60d22 100644 --- a/internal/tuic/server_test.go +++ b/internal/tuic/server_test.go @@ -10,12 +10,14 @@ import ( "crypto/x509" "crypto/x509/pkix" "encoding/pem" + "errors" "io" "math/big" "net" "testing" "time" + serverquic "github.com/apernet/quic-go" "github.com/google/uuid" "github.com/quic-go/quic-go" ) @@ -199,6 +201,27 @@ func testServerTCPConnectE2E(t *testing.T, controller string) { waitForClientTraffic(t, server, "alice@example.com", int64(len(testMsg))) } +// closeUniStream tolerates only the server's STOP_SENDING: it cancels the read side of a +// uni stream once the command is parsed, which can land before the client's FIN. +func closeUniStream(t *testing.T, stream interface { + Close() error + Context() context.Context +}, +) { + t.Helper() + err := stream.Close() + if err == nil { + return + } + cause := context.Cause(stream.Context()) + var clientErr *quic.StreamError + var serverErr *serverquic.StreamError + if (errors.As(cause, &clientErr) && clientErr.Remote) || (errors.As(cause, &serverErr) && serverErr.Remote) { + return + } + t.Fatalf("close uni stream: %v (cause %v)", err, cause) +} + // waitForClientTraffic accumulates drained deltas because the up and down counters are // bumped on different relay goroutines, so the echo can arrive before the upload is counted. func waitForClientTraffic(t *testing.T, server *Server, email string, minBytes int64) { @@ -445,9 +468,7 @@ func testServerUDPStreamE2E(t *testing.T, controller string) { if _, err := authStream.Write(authPayload); err != nil { t.Fatalf("write authentication payload failed: %v", err) } - if err := authStream.Close(); err != nil { - t.Fatalf("close authentication stream failed: %v", err) - } + closeUniStream(t, authStream) target := &Address{Type: AddrTypeIPv4, IP: net.ParseIP("8.8.8.8"), Port: 53} udpMsg := bytes.Repeat([]byte("s"), 8500) @@ -470,9 +491,7 @@ func testServerUDPStreamE2E(t *testing.T, controller string) { if _, err := packetStream.Write(frame.Bytes()); err != nil { t.Fatalf("write packet frame failed: %v", err) } - if err := packetStream.Close(); err != nil { - t.Fatalf("close packet stream failed: %v", err) - } + closeUniStream(t, packetStream) } replyReassembler := newPacketReassembler(maxUdpRelayPacketSize) diff --git a/internal/tuic/stream_udp_lifecycle_test.go b/internal/tuic/stream_udp_lifecycle_test.go index b3eec976e..94d5db092 100644 --- a/internal/tuic/stream_udp_lifecycle_test.go +++ b/internal/tuic/stream_udp_lifecycle_test.go @@ -57,9 +57,7 @@ func startLifecycleTestServer(t *testing.T, relayAddr, email string) (*Server, * if _, err := stream.Write(auth); err != nil { t.Fatalf("write authentication: %v", err) } - if err := stream.Close(); err != nil { - t.Fatalf("close authentication stream: %v", err) - } + closeUniStream(t, stream) return server, client, clientID, password }