quic/conn_streams_test.go

// Copyright 2023 The Go Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

//go:build go1.21

package quic

import (
	"context"
	"fmt"
	"io"
	"math"
	"sync"
	"testing"
)

func TestStreamsCreate(t *testing.T) {
	ctx := canceledContext()
	tc := newTestConn(t, clientSide, permissiveTransportParameters)
	tc.handshake()

	s, err := tc.conn.NewStream(ctx)
	if err != nil {
		t.Fatalf("NewStream: %v", err)
	}
	s.Flush() // open the stream
	tc.wantFrame("created bidirectional stream 0",
		packetType1RTT, debugFrameStream{
			id:   0, // client-initiated, bidi, number 0
			data: []byte{},
		})

	s, err = tc.conn.NewSendOnlyStream(ctx)
	if err != nil {
		t.Fatalf("NewStream: %v", err)
	}
	s.Flush() // open the stream
	tc.wantFrame("created unidirectional stream 0",
		packetType1RTT, debugFrameStream{
			id:   2, // client-initiated, uni, number 0
			data: []byte{},
		})

	s, err = tc.conn.NewStream(ctx)
	if err != nil {
		t.Fatalf("NewStream: %v", err)
	}
	s.Flush() // open the stream
	tc.wantFrame("created bidirectional stream 1",
		packetType1RTT, debugFrameStream{
			id:   4, // client-initiated, uni, number 4
			data: []byte{},
		})
}

func TestStreamsAccept(t *testing.T) {
	ctx := canceledContext()
	tc := newTestConn(t, serverSide)
	tc.handshake()

	tc.writeFrames(packetType1RTT,
		debugFrameStream{
			id: 0, // client-initiated, bidi, number 0
		},
		debugFrameStream{
			id: 2, // client-initiated, uni, number 0
		},
		debugFrameStream{
			id: 4, // client-initiated, bidi, number 1
		})

	for _, accept := range []struct {
		id       streamID
		readOnly bool
	}{
		{0, false},
		{2, true},
		{4, false},
	} {
		s, err := tc.conn.AcceptStream(ctx)
		if err != nil {
			t.Fatalf("conn.AcceptStream() = %v, want stream %v", err, accept.id)
		}
		if got, want := s.id, accept.id; got != want {
			t.Fatalf("conn.AcceptStream() = stream %v, want %v", got, want)
		}
		if got, want := s.IsReadOnly(), accept.readOnly; got != want {
			t.Fatalf("stream %v: s.IsReadOnly() = %v, want %v", accept.id, got, want)
		}
	}

	_, err := tc.conn.AcceptStream(ctx)
	if err != context.Canceled {
		t.Fatalf("conn.AcceptStream() = %v, want context.Canceled", err)
	}
}

func TestStreamsBlockingAccept(t *testing.T) {
	tc := newTestConn(t, serverSide)
	tc.handshake()

	a := runAsync(tc, func(ctx context.Context) (*Stream, error) {
		return tc.conn.AcceptStream(ctx)
	})
	if _, err := a.result(); err != errNotDone {
		tc.t.Fatalf("AcceptStream() = _, %v; want errNotDone", err)
	}

	sid := newStreamID(clientSide, bidiStream, 0)
	tc.writeFrames(packetType1RTT,
		debugFrameStream{
			id: sid,
		})

	s, err := a.result()
	if err != nil {
		t.Fatalf("conn.AcceptStream() = _, %v, want stream", err)
	}
	if got, want := s.id, sid; got != want {
		t.Fatalf("conn.AcceptStream() = stream %v, want %v", got, want)
	}
	if got, want := s.IsReadOnly(), false; got != want {
		t.Fatalf("s.IsReadOnly() = %v, want %v", got, want)
	}
}

func TestStreamsLocalStreamNotCreated(t *testing.T) {
	// "An endpoint MUST terminate the connection with error STREAM_STATE_ERROR
	// if it receives a STREAM frame for a locally initiated stream that has
	// not yet been created [...]"
	// https://www.rfc-editor.org/rfc/rfc9000.html#section-19.8-3
	tc := newTestConn(t, serverSide)
	tc.handshake()

	tc.writeFrames(packetType1RTT,
		debugFrameStream{
			id: 1, // server-initiated, bidi, number 0
		})
	tc.wantFrame("peer sent STREAM frame for an uncreated local stream",
		packetType1RTT, debugFrameConnectionCloseTransport{
			code: errStreamState,
		})
}

func TestStreamsLocalStreamClosed(t *testing.T) {
	tc, s := newTestConnAndLocalStream(t, clientSide, uniStream, permissiveTransportParameters)
	s.CloseWrite()
	tc.wantFrame("FIN for closed stream",
		packetType1RTT, debugFrameStream{
			id:   newStreamID(clientSide, uniStream, 0),
			fin:  true,
			data: []byte{},
		})
	tc.writeAckForAll()

	tc.writeFrames(packetType1RTT, debugFrameStopSending{
		id: newStreamID(clientSide, uniStream, 0),
	})
	tc.wantIdle("frame for finalized stream is ignored")

	// ACKing the last stream packet should have cleaned up the stream.
	// Check that we don't have any state left.
	if got := len(tc.conn.streams.streams); got != 0 {
		t.Fatalf("after close, len(tc.conn.streams.streams) = %v, want 0", got)
	}
	if tc.conn.streams.queueMeta.head != nil {
		t.Fatalf("after close, stream send queue is not empty; should be")
	}
}

func TestStreamsStreamSendOnly(t *testing.T) {
	// "An endpoint MUST terminate the connection with error STREAM_STATE_ERROR
	// if it receives a STREAM frame for a locally initiated stream that has
	// not yet been created [...]"
	// https://www.rfc-editor.org/rfc/rfc9000.html#section-19.8-3
	ctx := canceledContext()
	tc := newTestConn(t, serverSide, permissiveTransportParameters)
	tc.handshake()

	s, err := tc.conn.NewSendOnlyStream(ctx)
	if err != nil {
		t.Fatalf("NewStream: %v", err)
	}
	s.Flush() // open the stream
	tc.wantFrame("created unidirectional stream 0",
		packetType1RTT, debugFrameStream{
			id:   3, // server-initiated, uni, number 0
			data: []byte{},
		})

	tc.writeFrames(packetType1RTT,
		debugFrameStream{
			id: 3, // server-initiated, bidi, number 0
		})
	tc.wantFrame("peer sent STREAM frame for a send-only stream",
		packetType1RTT, debugFrameConnectionCloseTransport{
			code: errStreamState,
		})
}

func TestStreamsWriteQueueFairness(t *testing.T) {
	ctx := canceledContext()
	const dataLen = 1 << 20
	const numStreams = 3
	tc := newTestConn(t, clientSide, func(p *transportParameters) {
		p.initialMaxStreamsBidi = numStreams
		p.initialMaxData = 1<<62 - 1
		p.initialMaxStreamDataBidiRemote = dataLen
	}, func(c *Config) {
		c.MaxStreamWriteBufferSize = dataLen
	})
	tc.handshake()
	tc.ignoreFrame(frameTypeAck)

	// Create a number of streams, and write a bunch of data to them.
	// The streams are not limited by flow control.
	//
	// The first stream we create is going to immediately consume all
	// available congestion window.
	//
	// Once we've created all the remaining streams,
	// we start sending acks back to open up the congestion window.
	// We verify that all streams can make progress.
	data := make([]byte, dataLen)
	var streams []*Stream
	for i := 0; i < numStreams; i++ {
		s, err := tc.conn.NewStream(ctx)
		if err != nil {
			t.Fatal(err)
		}
		streams = append(streams, s)
		if n, err := s.Write(data); n != len(data) || err != nil {
			t.Fatalf("s.Write() = %v, %v; want %v, nil", n, err, len(data))
		}
		// Wait for the stream to finish writing whatever frames it can before
		// congestion control blocks it.
		tc.wait()
	}

	sent := make([]int64, len(streams))
	for {
		p := tc.readPacket()
		if p == nil {
			break
		}
		tc.writeFrames(packetType1RTT, debugFrameAck{
			ranges: []i64range[packetNumber]{{0, p.num}},
		})
		for _, f := range p.frames {
			sf, ok := f.(debugFrameStream)
			if !ok {
				t.Fatalf("got unexpected frame (want STREAM): %v", sf)
			}
			if got, want := sf.off, sent[sf.id.num()]; got != want {
				t.Fatalf("got frame: %v\nwant offset: %v", sf, want)
			}
			sent[sf.id.num()] = sf.off + int64(len(sf.data))
			// Look at the amount of data sent by all streams, excluding the first one.
			// (The first stream got a head start when it consumed the initial window.)
			//
			// We expect that difference between the streams making the most and least progress
			// so far will be less than the maximum datagram size.
			minSent := sent[1]
			maxSent := sent[1]
			for _, s := range sent[2:] {
				minSent = min(minSent, s)
				maxSent = max(maxSent, s)
			}
			const maxDelta = maxUDPPayloadSize
			if d := maxSent - minSent; d > maxDelta {
				t.Fatalf("stream data sent: %v; delta=%v, want delta <= %v", sent, d, maxDelta)
			}
		}
	}
	// Final check that every stream sent the full amount of data expected.
	for num, s := range sent {
		if s != dataLen {
			t.Errorf("stream %v sent %v bytes, want %v", num, s, dataLen)
		}
	}
}

func TestStreamsShutdown(t *testing.T) {
	// These tests verify that a stream is removed from the Conn's map of live streams
	// after it is fully shut down.
	//
	// Each case consists of a setup step, after which one stream should exist,
	// and a shutdown step, after which no streams should remain in the Conn.
	for _, test := range []struct {
		name     string
		side     streamSide
		styp     streamType
		setup    func(*testing.T, *testConn, *Stream)
		shutdown func(*testing.T, *testConn, *Stream)
	}{{
		name: "closed",
		side: localStream,
		styp: uniStream,
		setup: func(t *testing.T, tc *testConn, s *Stream) {
			s.Close()
		},
		shutdown: func(t *testing.T, tc *testConn, s *Stream) {
			tc.writeAckForAll()
		},
	}, {
		name: "local close",
		side: localStream,
		styp: bidiStream,
		setup: func(t *testing.T, tc *testConn, s *Stream) {
			tc.writeFrames(packetType1RTT, debugFrameResetStream{
				id: s.id,
			})
			s.Close()
		},
		shutdown: func(t *testing.T, tc *testConn, s *Stream) {
			tc.writeAckForAll()
		},
	}, {
		name: "remote reset",
		side: localStream,
		styp: bidiStream,
		setup: func(t *testing.T, tc *testConn, s *Stream) {
			s.Close()
			tc.wantIdle("all frames after Close are ignored")
			tc.writeAckForAll()
		},
		shutdown: func(t *testing.T, tc *testConn, s *Stream) {
			tc.writeFrames(packetType1RTT, debugFrameResetStream{
				id: s.id,
			})
		},
	}, {
		name: "local close",
		side: remoteStream,
		styp: uniStream,
		setup: func(t *testing.T, tc *testConn, s *Stream) {
			tc.writeFrames(packetType1RTT, debugFrameStream{
				id:  s.id,
				fin: true,
			})
			if n, err := s.Read(make([]byte, 16)); n != 0 || err != io.EOF {
				t.Errorf("Read() = %v, %v; want 0, io.EOF", n, err)
			}
		},
		shutdown: func(t *testing.T, tc *testConn, s *Stream) {
			s.CloseRead()
		},
	}} {
		name := fmt.Sprintf("%v/%v/%v", test.side, test.styp, test.name)
		t.Run(name, func(t *testing.T) {
			tc, s := newTestConnAndStream(t, serverSide, test.side, test.styp,
				permissiveTransportParameters)
			tc.ignoreFrame(frameTypeStreamBase)
			tc.ignoreFrame(frameTypeStopSending)
			test.setup(t, tc, s)
			tc.wantIdle("conn should be idle after setup")
			if got, want := len(tc.conn.streams.streams), 1; got != want {
				t.Fatalf("after setup: %v streams in Conn's map; want %v", got, want)
			}
			test.shutdown(t, tc, s)
			tc.wantIdle("conn should be idle after shutdown")
			if got, want := len(tc.conn.streams.streams), 0; got != want {
				t.Fatalf("after shutdown: %v streams in Conn's map; want %v", got, want)
			}
		})
	}
}

func TestStreamsCreateAndCloseRemote(t *testing.T) {
	// This test exercises creating new streams in response to frames
	// from the peer, and cleaning up after streams are fully closed.
	//
	// It's overfitted to the current implementation, but works through
	// a number of corner cases in that implementation.
	//
	// Disable verbose logging in this test: It sends a lot of packets,
	// and they're not especially interesting on their own.
	defer func(vv bool) {
		*testVV = vv
	}(*testVV)
	*testVV = false
	ctx := canceledContext()
	tc := newTestConn(t, serverSide, permissiveTransportParameters)
	tc.handshake()
	tc.ignoreFrame(frameTypeAck)
	type op struct {
		id streamID
	}
	type streamOp op
	type resetOp op
	type acceptOp op
	const noStream = math.MaxInt64
	stringID := func(id streamID) string {
		return fmt.Sprintf("%v/%v", id.streamType(), id.num())
	}
	for _, op := range []any{
		"opening bidi/5 implicitly opens bidi/0-4",
		streamOp{newStreamID(clientSide, bidiStream, 5)},
		acceptOp{newStreamID(clientSide, bidiStream, 5)},
		"bidi/3 was implicitly opened",
		streamOp{newStreamID(clientSide, bidiStream, 3)},
		acceptOp{newStreamID(clientSide, bidiStream, 3)},
		resetOp{newStreamID(clientSide, bidiStream, 3)},
		"bidi/3 is done, frames for it are discarded",
		streamOp{newStreamID(clientSide, bidiStream, 3)},
		"open and close some uni streams as well",
		streamOp{newStreamID(clientSide, uniStream, 0)},
		acceptOp{newStreamID(clientSide, uniStream, 0)},
		streamOp{newStreamID(clientSide, uniStream, 1)},
		acceptOp{newStreamID(clientSide, uniStream, 1)},
		streamOp{newStreamID(clientSide, uniStream, 2)},
		acceptOp{newStreamID(clientSide, uniStream, 2)},
		resetOp{newStreamID(clientSide, uniStream, 1)},
		resetOp{newStreamID(clientSide, uniStream, 0)},
		resetOp{newStreamID(clientSide, uniStream, 2)},
		"closing an implicitly opened stream causes us to accept it",
		resetOp{newStreamID(clientSide, bidiStream, 0)},
		acceptOp{newStreamID(clientSide, bidiStream, 0)},
		resetOp{newStreamID(clientSide, bidiStream, 1)},
		acceptOp{newStreamID(clientSide, bidiStream, 1)},
		resetOp{newStreamID(clientSide, bidiStream, 2)},
		acceptOp{newStreamID(clientSide, bidiStream, 2)},
		"stream bidi/3 was reset previously",
		resetOp{newStreamID(clientSide, bidiStream, 3)},
		resetOp{newStreamID(clientSide, bidiStream, 4)},
		acceptOp{newStreamID(clientSide, bidiStream, 4)},
		"stream bidi/5 was reset previously",
		resetOp{newStreamID(clientSide, bidiStream, 5)},
		"stream bidi/6 was not implicitly opened",
		resetOp{newStreamID(clientSide, bidiStream, 6)},
		acceptOp{newStreamID(clientSide, bidiStream, 6)},
	} {
		if _, ok := op.(acceptOp); !ok {
			if s, err := tc.conn.AcceptStream(ctx); err == nil {
				t.Fatalf("accepted stream %v, want none", stringID(s.id))
			}
		}
		switch op := op.(type) {
		case string:
			t.Log("# " + op)
		case streamOp:
			t.Logf("open stream %v", stringID(op.id))
			tc.writeFrames(packetType1RTT, debugFrameStream{
				id: streamID(op.id),
			})
		case resetOp:
			t.Logf("reset stream %v", stringID(op.id))
			tc.writeFrames(packetType1RTT, debugFrameResetStream{
				id: op.id,
			})
		case acceptOp:
			s := tc.acceptStream()
			if s.id != op.id {
				t.Fatalf("accepted stream %v; want stream %v", stringID(s.id), stringID(op.id))
			}
			t.Logf("accepted stream %v", stringID(op.id))
			// Immediately close the stream, so the stream becomes done when the
			// peer closes its end.
			s.Close()
		}
		p := tc.readPacket()
		if p != nil {
			tc.writeFrames(p.ptype, debugFrameAck{
				ranges: []i64range[packetNumber]{{0, p.num + 1}},
			})
		}
	}
	// Every stream should be fully closed now.
	// Check that we don't have any state left.
	if got := len(tc.conn.streams.streams); got != 0 {
		t.Fatalf("after test, len(tc.conn.streams.streams) = %v, want 0", got)
	}
	if tc.conn.streams.queueMeta.head != nil {
		t.Fatalf("after test, stream send queue is not empty; should be")
	}
}

func TestStreamsCreateConcurrency(t *testing.T) {
	cli, srv := newLocalConnPair(t, &Config{}, &Config{})

	srvdone := make(chan int)
	go func() {
		defer close(srvdone)
		for streams := 0; ; streams++ {
			s, err := srv.AcceptStream(context.Background())
			if err != nil {
				srvdone <- streams
				return
			}
			s.Close()
		}
	}()

	var wg sync.WaitGroup
	const concurrency = 10
	const streams = 10
	for i := 0; i < concurrency; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := 0; j < streams; j++ {
				s, err := cli.NewStream(context.Background())
				if err != nil {
					t.Errorf("NewStream: %v", err)
					return
				}
				s.Flush()
				_, err = io.ReadAll(s)
				if err != nil {
					t.Errorf("ReadFull: %v", err)
				}
				s.Close()
			}
		}()
	}
	wg.Wait()

	cli.Abort(nil)
	srv.Abort(nil)
	if got, want := <-srvdone, concurrency*streams; got != want {
		t.Errorf("accepted %v streams, want %v", got, want)
	}
}

func TestStreamsPTOWithImplicitStream(t *testing.T) {
	ctx := canceledContext()
	tc := newTestConn(t, serverSide, permissiveTransportParameters)
	tc.handshake()
	tc.ignoreFrame(frameTypeAck)

	// Peer creates stream 1, and implicitly creates stream 0.
	tc.writeFrames(packetType1RTT, debugFrameStream{
		id: newStreamID(clientSide, bidiStream, 1),
	})

	// We accept stream 1 and write data to it.
	data := []byte("data")
	s, err := tc.conn.AcceptStream(ctx)
	if err != nil {
		t.Fatalf("conn.AcceptStream() = %v, want stream", err)
	}
	s.Write(data)
	s.Flush()
	tc.wantFrame("data written to stream",
		packetType1RTT, debugFrameStream{
			id:   newStreamID(clientSide, bidiStream, 1),
			data: data,
		})

	// PTO expires, and the data is resent.
	const pto = true
	tc.triggerLossOrPTO(packetType1RTT, true)
	tc.wantFrame("data resent after PTO expires",
		packetType1RTT, debugFrameStream{
			id:   newStreamID(clientSide, bidiStream, 1),
			data: data,
		})
}