add proper tests for the session manager

Author: marten-seemann

.github/workflows/lint.yml

@@ -11,6 +11,8 @@ jobs:
         go: [ "1.24.x", "1.25.x" ]
     runs-on: ubuntu-latest
     name: Lint
+    env:
+      GOEXPERIMENT: ${{ matrix.go == '1.24.x' && 'synctest' || '' }}
     steps:
       - uses: actions/checkout@v6
       - uses: actions/setup-go@v6

.github/workflows/unit.yml

@@ -12,6 +12,8 @@ jobs:
         go: [ "1.24.x", "1.25.x" ]
     runs-on: ${{ fromJSON(vars[format('UNIT_RUNNER_{0}', matrix.os)] || format('"{0}-latest"', matrix.os)) }}
     name: Unit tests (${{ matrix.os}}, Go ${{ matrix.go }})
+    env:
+      GOEXPERIMENT: ${{ matrix.go == '1.24.x' && 'synctest' || '' }}
     steps:
       - uses: actions/checkout@v6
       - uses: actions/setup-go@v6

internal/synctest/synctest_go124.go

@@ -0,0 +1,19 @@
+//go:build go1.24 && !go1.25
+
+package synctest
+
+import (
+	"testing"
+	"testing/synctest"
+)
+
+func Test(t *testing.T, f func(t *testing.T)) {
+	synctest.Run(func() {
+		f(t)
+	})
+}
+
+func Wait() {
+	//nolint:govet // the CI configuration sets the GOEXPERIMENT=synctest flag
+	synctest.Wait()
+}

internal/synctest/synctest_go125.go

@@ -0,0 +1,16 @@
+//go:build go1.25
+
+package synctest
+
+import (
+	"testing"
+	"testing/synctest"
+)
+
+func Test(t *testing.T, f func(t *testing.T)) {
+	synctest.Test(t, f)
+}
+
+func Wait() {
+	synctest.Wait()
+}

server_test.go

@@ -188,88 +188,6 @@ func TestServerReorderedUpgradeRequestTimeout(t *testing.T) {
 	require.Equal(t, []byte("raboof"), data)
 }
 
-func TestServerReorderedMultipleStreams(t *testing.T) {
-	timeout := scaleDuration(150 * time.Millisecond)
-	s := webtransport.Server{
-		H3:                &http3.Server{TLSConfig: webtransport.TLSConf, EnableDatagrams: true},
-		ReorderingTimeout: timeout,
-	}
-	defer s.Close()
-	connChan := make(chan *webtransport.Session)
-	addHandler(t, &s, func(c *webtransport.Session) {
-		connChan <- c
-	})
-
-	udpConn, err := net.ListenUDP("udp", nil)
-	require.NoError(t, err)
-	port := udpConn.LocalAddr().(*net.UDPAddr).Port
-	webtransport.ConfigureHTTP3Server(s.H3)
-	go s.Serve(udpConn)
-
-	cconn, err := quic.DialAddr(
-		context.Background(),
-		fmt.Sprintf("localhost:%d", port),
-		&tls.Config{RootCAs: webtransport.CertPool, NextProtos: []string{http3.NextProtoH3}},
-		&quic.Config{EnableDatagrams: true},
-	)
-	require.NoError(t, err)
-	start := time.Now()
-	// Open a new stream for a WebTransport session we'll establish later.
-	str1 := createStreamAndWrite(t, cconn, 8, []byte("foobar"))
-
-	// After a while, open another stream for the same session.
-	// This resets the timer, so the timeout will be timeout after this point.
-	time.Sleep(timeout / 2)
-	str2 := createStreamAndWrite(t, cconn, 8, []byte("raboof"))
-
-	// Wait for timeout after the second stream was added.
-	// The timer was reset when str2 was added, so both streams should be reset
-	// timeout after str2 was added, which is timeout/2 + timeout = 1.5*timeout from start.
-	time.Sleep(timeout + timeout/4)
-
-	// Both streams should now have been reset by the server.
-	_, err = str1.Read([]byte{0})
-	var streamErr *quic.StreamError
-	require.ErrorAs(t, err, &streamErr)
-	require.Equal(t, webtransport.WTBufferedStreamRejectedErrorCode, streamErr.ErrorCode)
-
-	_, err = str2.Read([]byte{0})
-	require.ErrorAs(t, err, &streamErr)
-	require.Equal(t, webtransport.WTBufferedStreamRejectedErrorCode, streamErr.ErrorCode)
-
-	took := time.Since(start)
-	require.GreaterOrEqual(t, took, timeout*3/2)
-	require.Less(t, took, timeout*2)
-
-	tr := &http3.Transport{EnableDatagrams: true}
-	conn := tr.NewClientConn(cconn)
-	// Now establish the session. Make sure we don't accept the streams (they were reset).
-	requestStr, err := conn.OpenRequestStream(context.Background())
-	require.NoError(t, err)
-	require.NoError(t, requestStr.SendRequestHeader(
-		webtransport.NewWebTransportRequest(t, fmt.Sprintf("https://localhost:%d/webtransport", port)),
-	))
-	rsp, err := requestStr.ReadResponse()
-	require.NoError(t, err)
-	require.Equal(t, http.StatusOK, rsp.StatusCode)
-	sconn := <-connChan
-	defer sconn.CloseWithError(0, "")
-	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
-	defer cancel()
-	_, err = sconn.AcceptStream(ctx)
-	require.ErrorIs(t, err, context.DeadlineExceeded)
-
-	// Establish another stream and make sure it's accepted now.
-	createStreamAndWrite(t, cconn, 8, []byte("baz"))
-	ctx, cancel = context.WithTimeout(context.Background(), 200*time.Millisecond)
-	defer cancel()
-	sstr, err := sconn.AcceptStream(ctx)
-	require.NoError(t, err)
-	data, err := io.ReadAll(sstr)
-	require.NoError(t, err)
-	require.Equal(t, []byte("baz"), data)
-}
-
 func TestServerSettingsCheck(t *testing.T) {
 	timeout := scaleDuration(150 * time.Millisecond)
 	s := webtransport.Server{

session_manager_test.go

@@ -3,18 +3,46 @@ package webtransport
 import (
 	"context"
 	"io"
+	"net"
 	"net/http"
 	"net/http/httptest"
 	"testing"
 	"time"
 
+	"github.com/quic-go/webtransport-go/internal/synctest"
+
+	"github.com/quic-go/quic-go"
 	"github.com/quic-go/quic-go/http3"
+	"github.com/quic-go/quic-go/testutils/simnet"
 
 	"github.com/stretchr/testify/require"
 )
 
-func TestSessionManager(t *testing.T) {
-	clientConn, serverConn := newConnPair(t)
+func (m *sessionManager) NumSessions() int {
+	m.mx.Lock()
+	defer m.mx.Unlock()
+	return len(m.sessions)
+}
+
+func newSimnetLink(t *testing.T, rtt time.Duration) (client, server *simnet.SimConn, close func(t *testing.T)) {
+	t.Helper()
+
+	n := &simnet.Simnet{Router: &simnet.PerfectRouter{}}
+	settings := simnet.NodeBiDiLinkSettings{Latency: rtt / 2}
+	clientPacketConn := n.NewEndpoint(&net.UDPAddr{IP: net.ParseIP("1.0.0.1"), Port: 9001}, settings)
+	serverPacketConn := n.NewEndpoint(&net.UDPAddr{IP: net.ParseIP("1.0.0.2"), Port: 9002}, settings)
+
+	require.NoError(t, n.Start())
+
+	return clientPacketConn, serverPacketConn, func(t *testing.T) {
+		require.NoError(t, clientPacketConn.Close())
+		require.NoError(t, serverPacketConn.Close())
+		require.NoError(t, n.Close())
+	}
+}
+
+func TestSessionManagerAddingStreams(t *testing.T) {
+	clientConn, serverConn := newConnPair(t, newUDPConnLocalhost(t), newUDPConnLocalhost(t))
 	t.Cleanup(func() {
 		clientConn.CloseWithError(0, "")
 		serverConn.CloseWithError(0, "")
@@ -44,18 +72,34 @@ func TestSessionManager(t *testing.T) {
 	reqStr, err := cc.OpenRequestStream(ctx)
 	require.NoError(t, err)
 	require.NoError(t, reqStr.SendRequestHeader(httptest.NewRequest(http.MethodGet, "/", nil)))
-	// TODO: somehow send an HTTP response so we can call ReadResponse()
-	reqStr.ReadResponse()
+	server := http3.Server{
+		Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			w.WriteHeader(http.StatusOK)
+			w.(http.Flusher).Flush()
+			<-r.Context().Done()
+		}),
+	}
+	sc, err := server.NewRawServerConn(serverConn)
+	require.NoError(t, err)
+	serverReqStr, err := serverConn.AcceptStream(ctx)
+	require.NoError(t, err)
+	go sc.HandleRequestStream(serverReqStr)
+
+	_, err = reqStr.ReadResponse()
+	require.NoError(t, err)
 
 	const sessionID = 42
 
 	sessMgr := newSessionManager(time.Hour)
+	require.Zero(t, sessMgr.NumSessions())
 	// first add the streams...
+	sess := newSession(context.Background(), sessionID, clientConn, reqStr, "")
+	// ...then add the session
 	sessMgr.AddStream(clientStr, sessionID)
+	require.Equal(t, 1, sessMgr.NumSessions())
 	sessMgr.AddUniStream(clientUniStr, sessionID)
-	// ...then add the session
-	sess := newSession(context.Background(), sessionID, clientConn, reqStr, "")
 	sessMgr.AddSession(sessionID, sess)
+	require.Equal(t, 1, sessMgr.NumSessions())
 
 	// the streams should now be returned from the session
 	sessStr, err := sess.AcceptStream(ctx)
@@ -70,3 +114,123 @@ func TestSessionManager(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, []byte("world"), data)
 }
+
+func TestSessionManagerStreamReordering(t *testing.T) {
+	synctest.Test(t, func(t *testing.T) {
+		clientPacketConn, serverPacketConn, closeFn := newSimnetLink(t, 10*time.Millisecond)
+		defer closeFn(t)
+		clientConn, serverConn := newConnPair(t, clientPacketConn, serverPacketConn)
+		t.Cleanup(func() {
+			clientConn.CloseWithError(0, "")
+			serverConn.CloseWithError(0, "")
+		})
+
+		serverStr1, err := serverConn.OpenStream()
+		require.NoError(t, err)
+		_, err = serverStr1.Write([]byte("lorem"))
+		require.NoError(t, err)
+		require.NoError(t, serverStr1.Close())
+
+		serverUniStr1, err := serverConn.OpenUniStream()
+		require.NoError(t, err)
+		_, err = serverUniStr1.Write([]byte("ipsum"))
+		require.NoError(t, err)
+		require.NoError(t, serverUniStr1.Close())
+
+		serverStr2, err := serverConn.OpenStream()
+		require.NoError(t, err)
+		_, err = serverStr2.Write([]byte("dolor"))
+		require.NoError(t, err)
+		require.NoError(t, serverStr2.Close())
+
+		serverUniStr2, err := serverConn.OpenUniStream()
+		require.NoError(t, err)
+		_, err = serverUniStr2.Write([]byte("sit"))
+		require.NoError(t, err)
+		require.NoError(t, serverUniStr2.Close())
+
+		ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+		defer cancel()
+		clientStr1, err := clientConn.AcceptStream(ctx)
+		require.NoError(t, err)
+		clientUniStr1, err := clientConn.AcceptUniStream(ctx)
+		require.NoError(t, err)
+		clientStr2, err := clientConn.AcceptStream(ctx)
+		require.NoError(t, err)
+		clientUniStr2, err := clientConn.AcceptUniStream(ctx)
+		require.NoError(t, err)
+
+		tr := http3.Transport{}
+		cc := tr.NewRawClientConn(clientConn)
+		reqStr, err := cc.OpenRequestStream(ctx)
+		require.NoError(t, err)
+		require.NoError(t, reqStr.SendRequestHeader(httptest.NewRequest(http.MethodGet, "/", nil)))
+		server := http3.Server{
+			Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(http.StatusOK)
+				w.(http.Flusher).Flush()
+				<-r.Context().Done()
+			}),
+		}
+		sc, err := server.NewRawServerConn(serverConn)
+		require.NoError(t, err)
+		serverReqStr, err := serverConn.AcceptStream(ctx)
+		require.NoError(t, err)
+		go sc.HandleRequestStream(serverReqStr)
+
+		_, err = reqStr.ReadResponse()
+		require.NoError(t, err)
+
+		const sessionID = 42
+		const timeout = 3 * time.Second
+
+		sessMgr := newSessionManager(timeout)
+		require.Zero(t, sessMgr.NumSessions())
+		// add the first stream
+		sessMgr.AddStream(clientStr1, sessionID)
+		require.Equal(t, 1, sessMgr.NumSessions())
+		time.Sleep(timeout + time.Second)
+		// the stream should have been reset and the session manager should have no sessions
+		require.Zero(t, sessMgr.NumSessions())
+		_, err = serverStr1.Read([]byte{0})
+		require.ErrorIs(t, err, &quic.StreamError{Remote: true, StreamID: serverStr1.StreamID(), ErrorCode: WTBufferedStreamRejectedErrorCode})
+
+		sessMgr.AddUniStream(clientUniStr1, sessionID)
+		require.Equal(t, 1, sessMgr.NumSessions())
+		time.Sleep(timeout - time.Second)
+		// adding another stream resets the timer
+		sessMgr.AddStream(clientStr2, sessionID)
+		time.Sleep(timeout - time.Second)
+		require.Equal(t, 1, sessMgr.NumSessions())
+
+		// now add the session
+		sess := newSession(context.Background(), sessionID, clientConn, reqStr, "")
+		sessMgr.AddSession(sessionID, sess)
+		require.Equal(t, 1, sessMgr.NumSessions())
+
+		// wait for a long time, then add another stream
+		time.Sleep(timeout + time.Second)
+		require.Equal(t, 1, sessMgr.NumSessions())
+		sessMgr.AddUniStream(clientUniStr2, sessionID)
+		time.Sleep(timeout + time.Second)
+
+		// the "lorem" stream should have been reset and the "dolor" stream should have been returned
+		sessStr, err := sess.AcceptStream(ctx)
+		require.NoError(t, err)
+		data, err := io.ReadAll(sessStr)
+		require.NoError(t, err)
+		require.Equal(t, []byte("dolor"), data)
+
+		sessUniStr1, err := sess.AcceptUniStream(ctx)
+		require.NoError(t, err)
+		data, err = io.ReadAll(sessUniStr1)
+		require.NoError(t, err)
+		require.Equal(t, []byte("ipsum"), data)
+
+		sessUniStr2, err := sess.AcceptUniStream(ctx)
+		require.NoError(t, err)
+		data, err = io.ReadAll(sessUniStr2)
+		require.NoError(t, err)
+		require.Equal(t, []byte("sit"), data)
+	})
+}