Repository: quic-go/quic-go
Pull request #4844 (2025-01-07 06:31:21)

reliably queue MAX_DATA frames
We need to make sure that a MAX_DATA frame actually triggers the packing of a new packet (with which it can then be sent out). Otherwise, reading from a stream might get the MAX_DATA frame queued, but if no packet is sent, it's never sent out to the peer, potentially leading to a deadlock of the connection.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) Report
Attention: Patch coverage is `87.09677%` with `4 lines` in your changes missing coverage. Please review.
> Project coverage is 83.52%. Comparing base [(`3cb5f3e`)](https://app.codecov.io/gh/quic-go/quic-go/commit/3cb5f3e104d269768415cce79ddcc5018c79ea92?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) to head [(`3c72700`)](https://app.codecov.io/gh/quic-go/quic-go/commit/3c727001ce9ed9608e368c2b79fdba0b54f23c75?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).

| [Files with missing lines](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) | Patch % | Lines |
|---|---|---|
| [receive\_stream.go](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?src=pr&el=tree&filepath=receive_stream.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go#diff-cmVjZWl2ZV9zdHJlYW0uZ28=) | 86.96% | [3 Missing :warning: ](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) |
| [connection.go](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?src=pr&el=tree&filepath=connection.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go#diff-Y29ubmVjdGlvbi5nbw==) | 0.00% | [1 Missing :warning: ](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) |

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master    #4844      +/-   ##
==========================================
- Coverage   83.69%   83.52%   -0.18%     
==========================================
  Files         149      146       -3     
  Lines       16078    15742     -336     
==========================================
- Hits        13456    13147     -309     
+ Misses       2091     2083       -8     
+ Partials      531      512      -19     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/quic-go/quic-go/pull/4844?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).   
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).


Patch:
diff --git a/connection.go b/connection.go
index 583e9a90..3371b76a 100644
--- a/connection.go
+++ b/connection.go
@@ -2282,6 +2282,8 @@ func (s *connection) queueControlFrame(f wire.Frame) {
 	s.scheduleSending()
 }
 
+func (s *connection) onHasConnectionData() { s.scheduleSending() }
+
 func (s *connection) onHasStreamData(id protocol.StreamID, str sendStreamI) {
 	s.framer.AddActiveStream(id, str)
 	s.scheduleSending()
diff --git a/internal/flowcontrol/connection_flow_controller.go b/internal/flowcontrol/connection_flow_controller.go
index 362a9a95..bbeb7889 100644
--- a/internal/flowcontrol/connection_flow_controller.go
+++ b/internal/flowcontrol/connection_flow_controller.go
@@ -57,10 +57,12 @@ func (c *connectionFlowController) IncrementHighestReceived(increment protocol.B
 	return nil
 }
 
-func (c *connectionFlowController) AddBytesRead(n protocol.ByteCount) {
+func (c *connectionFlowController) AddBytesRead(n protocol.ByteCount) (hasWindowUpdate bool) {
 	c.mutex.Lock()
+	defer c.mutex.Unlock()
+
 	c.baseFlowController.addBytesRead(n)
-	c.mutex.Unlock()
+	return c.baseFlowController.hasWindowUpdate()
 }
 
 func (c *connectionFlowController) GetWindowUpdate(now time.Time) protocol.ByteCount {
diff --git a/internal/flowcontrol/interface.go b/internal/flowcontrol/interface.go
index be641de1..23cf30c5 100644
--- a/internal/flowcontrol/interface.go
+++ b/internal/flowcontrol/interface.go
@@ -18,7 +18,7 @@ type flowController interface {
 // A StreamFlowController is a flow controller for a QUIC stream.
 type StreamFlowController interface {
 	flowController
-	AddBytesRead(protocol.ByteCount) (shouldQueueWindowUpdate bool)
+	AddBytesRead(protocol.ByteCount) (hasStreamWindowUpdate, hasConnWindowUpdate bool)
 	// UpdateHighestReceived is called when a new highest offset is received
 	// final has to be to true if this is the final offset of the stream,
 	// as contained in a STREAM frame with FIN bit, and the RESET_STREAM frame
@@ -32,7 +32,7 @@ type StreamFlowController interface {
 // The ConnectionFlowController is the flow controller for the connection.
 type ConnectionFlowController interface {
 	flowController
-	AddBytesRead(protocol.ByteCount)
+	AddBytesRead(protocol.ByteCount) (hasWindowUpdate bool)
 	Reset() error
 	IsNewlyBlocked() (bool, protocol.ByteCount)
 }
diff --git a/internal/flowcontrol/stream_flow_controller.go b/internal/flowcontrol/stream_flow_controller.go
index e12f1f46..ba005122 100644
--- a/internal/flowcontrol/stream_flow_controller.go
+++ b/internal/flowcontrol/stream_flow_controller.go
@@ -98,12 +98,12 @@ func (c *streamFlowController) UpdateHighestReceived(offset protocol.ByteCount,
 	return c.connection.IncrementHighestReceived(increment, now)
 }
 
-func (c *streamFlowController) AddBytesRead(n protocol.ByteCount) (shouldQueueWindowUpdate bool) {
+func (c *streamFlowController) AddBytesRead(n protocol.ByteCount) (hasStreamWindowUpdate, hasConnWindowUpdate bool) {
 	c.mutex.Lock()
 	c.baseFlowController.addBytesRead(n)
-	shouldQueueWindowUpdate = c.shouldQueueWindowUpdate()
+	hasStreamWindowUpdate = c.shouldQueueWindowUpdate()
 	c.mutex.Unlock()
-	c.connection.AddBytesRead(n)
+	hasConnWindowUpdate = c.connection.AddBytesRead(n)
 	return
 }
 
diff --git a/internal/mocks/connection_flow_controller.go b/internal/mocks/connection_flow_controller.go
index face2b21..19a71d13 100644
--- a/internal/mocks/connection_flow_controller.go
+++ b/internal/mocks/connection_flow_controller.go
@@ -42,9 +42,11 @@ func (m *MockConnectionFlowController) EXPECT() *MockConnectionFlowControllerMoc
 }
 
 // AddBytesRead mocks base method.
-func (m *MockConnectionFlowController) AddBytesRead(arg0 protocol.ByteCount) {
+func (m *MockConnectionFlowController) AddBytesRead(arg0 protocol.ByteCount) bool {
 	m.ctrl.T.Helper()
-	m.ctrl.Call(m, "AddBytesRead", arg0)
+	ret := m.ctrl.Call(m, "AddBytesRead", arg0)
+	ret0, _ := ret[0].(bool)
+	return ret0
 }
 
 // AddBytesRead indicates an expected call of AddBytesRead.
@@ -60,19 +62,19 @@ type MockConnectionFlowControllerAddBytesReadCall struct {
 }
 
 // Return rewrite *gomock.Call.Return
-func (c *MockConnectionFlowControllerAddBytesReadCall) Return() *MockConnectionFlowControllerAddBytesReadCall {
-	c.Call = c.Call.Return()
+func (c *MockConnectionFlowControllerAddBytesReadCall) Return(hasWindowUpdate bool) *MockConnectionFlowControllerAddBytesReadCall {
+	c.Call = c.Call.Return(hasWindowUpdate)
 	return c
 }
 
 // Do rewrite *gomock.Call.Do
-func (c *MockConnectionFlowControllerAddBytesReadCall) Do(f func(protocol.ByteCount)) *MockConnectionFlowControllerAddBytesReadCall {
+func (c *MockConnectionFlowControllerAddBytesReadCall) Do(f func(protocol.ByteCount) bool) *MockConnectionFlowControllerAddBytesReadCall {
 	c.Call = c.Call.Do(f)
 	return c
 }
 
 // DoAndReturn rewrite *gomock.Call.DoAndReturn
-func (c *MockConnectionFlowControllerAddBytesReadCall) DoAndReturn(f func(protocol.ByteCount)) *MockConnectionFlowControllerAddBytesReadCall {
+func (c *MockConnectionFlowControllerAddBytesReadCall) DoAndReturn(f func(protocol.ByteCount) bool) *MockConnectionFlowControllerAddBytesReadCall {
 	c.Call = c.Call.DoAndReturn(f)
 	return c
 }
diff --git a/internal/mocks/stream_flow_controller.go b/internal/mocks/stream_flow_controller.go
index 8a116787..38216131 100644
--- a/internal/mocks/stream_flow_controller.go
+++ b/internal/mocks/stream_flow_controller.go
@@ -78,11 +78,12 @@ func (c *MockStreamFlowControllerAbandonCall) DoAndReturn(f func()) *MockStreamF
 }
 
 // AddBytesRead mocks base method.
-func (m *MockStreamFlowController) AddBytesRead(arg0 protocol.ByteCount) bool {
+func (m *MockStreamFlowController) AddBytesRead(arg0 protocol.ByteCount) (bool, bool) {
 	m.ctrl.T.Helper()
 	ret := m.ctrl.Call(m, "AddBytesRead", arg0)
 	ret0, _ := ret[0].(bool)
-	return ret0
+	ret1, _ := ret[1].(bool)
+	return ret0, ret1
 }
 
 // AddBytesRead indicates an expected call of AddBytesRead.
@@ -98,19 +99,19 @@ type MockStreamFlowControllerAddBytesReadCall struct {
 }
 
 // Return rewrite *gomock.Call.Return
-func (c *MockStreamFlowControllerAddBytesReadCall) Return(shouldQueueWindowUpdate bool) *MockStreamFlowControllerAddBytesReadCall {
-	c.Call = c.Call.Return(shouldQueueWindowUpdate)
+func (c *MockStreamFlowControllerAddBytesReadCall) Return(hasStreamWindowUpdate, hasConnWindowUpdate bool) *MockStreamFlowControllerAddBytesReadCall {
+	c.Call = c.Call.Return(hasStreamWindowUpdate, hasConnWindowUpdate)
 	return c
 }
 
 // Do rewrite *gomock.Call.Do
-func (c *MockStreamFlowControllerAddBytesReadCall) Do(f func(protocol.ByteCount) bool) *MockStreamFlowControllerAddBytesReadCall {
+func (c *MockStreamFlowControllerAddBytesReadCall) Do(f func(protocol.ByteCount) (bool, bool)) *MockStreamFlowControllerAddBytesReadCall {
 	c.Call = c.Call.Do(f)
 	return c
 }
 
 // DoAndReturn rewrite *gomock.Call.DoAndReturn
-func (c *MockStreamFlowControllerAddBytesReadCall) DoAndReturn(f func(protocol.ByteCount) bool) *MockStreamFlowControllerAddBytesReadCall {
+func (c *MockStreamFlowControllerAddBytesReadCall) DoAndReturn(f func(protocol.ByteCount) (bool, bool)) *MockStreamFlowControllerAddBytesReadCall {
 	c.Call = c.Call.DoAndReturn(f)
 	return c
 }
diff --git a/receive_stream.go b/receive_stream.go
index 9b1b71cb..192b92f7 100644
--- a/receive_stream.go
+++ b/receive_stream.go
@@ -91,16 +91,19 @@ func (s *receiveStream) Read(p []byte) (int, error) {
 	defer func() { <-s.readOnce }()
 
 	s.mutex.Lock()
-	queuedNewControlFrame, n, err := s.readImpl(p)
+	queuedStreamWindowUpdate, queuedConnWindowUpdate, n, err := s.readImpl(p)
 	completed := s.isNewlyCompleted()
 	s.mutex.Unlock()
 
 	if completed {
 		s.sender.onStreamCompleted(s.streamID)
 	}
-	if queuedNewControlFrame {
+	if queuedStreamWindowUpdate {
 		s.sender.onHasStreamControlFrame(s.streamID, s)
 	}
+	if queuedConnWindowUpdate {
+		s.sender.onHasConnectionData()
+	}
 	return n, err
 }
 
@@ -125,20 +128,19 @@ func (s *receiveStream) isNewlyCompleted() bool {
 	return false
 }
 
-func (s *receiveStream) readImpl(p []byte) (bool, int, error) {
+func (s *receiveStream) readImpl(p []byte) (hasStreamWindowUpdate bool, hasConnWindowUpdate bool, _ int, _ error) {
 	if s.currentFrameIsLast && s.currentFrame == nil {
 		s.errorRead = true
-		return false, 0, io.EOF
+		return false, false, 0, io.EOF
 	}
 	if s.cancelledRemotely || s.cancelledLocally {
 		s.errorRead = true
-		return false, 0, s.cancelErr
+		return false, false, 0, s.cancelErr
 	}
 	if s.closeForShutdownErr != nil {
-		return false, 0, s.closeForShutdownErr
+		return false, false, 0, s.closeForShutdownErr
 	}
 
-	var queuedNewControlFrame bool
 	var bytesRead int
 	var deadlineTimer *utils.Timer
 	for bytesRead < len(p) {
@@ -146,23 +148,23 @@ func (s *receiveStream) readImpl(p []byte) (bool, int, error) {
 			s.dequeueNextFrame()
 		}
 		if s.currentFrame == nil && bytesRead > 0 {
-			return queuedNewControlFrame, bytesRead, s.closeForShutdownErr
+			return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, s.closeForShutdownErr
 		}
 
 		for {
 			// Stop waiting on errors
 			if s.closeForShutdownErr != nil {
-				return queuedNewControlFrame, bytesRead, s.closeForShutdownErr
+				return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, s.closeForShutdownErr
 			}
 			if s.cancelledRemotely || s.cancelledLocally {
 				s.errorRead = true
-				return queuedNewControlFrame, 0, s.cancelErr
+				return hasStreamWindowUpdate, hasConnWindowUpdate, 0, s.cancelErr
 			}
 
 			deadline := s.deadline
 			if !deadline.IsZero() {
 				if !time.Now().Before(deadline) {
-					return queuedNewControlFrame, bytesRead, errDeadline
+					return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, errDeadline
 				}
 				if deadlineTimer == nil {
 					deadlineTimer = utils.NewTimer()
@@ -192,10 +194,10 @@ func (s *receiveStream) readImpl(p []byte) (bool, int, error) {
 		}
 
 		if bytesRead > len(p) {
-			return queuedNewControlFrame, bytesRead, fmt.Errorf("BUG: bytesRead (%d) > len(p) (%d) in stream.Read", bytesRead, len(p))
+			return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, fmt.Errorf("BUG: bytesRead (%d) > len(p) (%d) in stream.Read", bytesRead, len(p))
 		}
 		if s.readPosInFrame > len(s.currentFrame) {
-			return queuedNewControlFrame, bytesRead, fmt.Errorf("BUG: readPosInFrame (%d) > frame.DataLen (%d) in stream.Read", s.readPosInFrame, len(s.currentFrame))
+			return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, fmt.Errorf("BUG: readPosInFrame (%d) > frame.DataLen (%d) in stream.Read", s.readPosInFrame, len(s.currentFrame))
 		}
 
 		m := copy(p[bytesRead:], s.currentFrame[s.readPosInFrame:])
@@ -205,9 +207,13 @@ func (s *receiveStream) readImpl(p []byte) (bool, int, error) {
 		// when a RESET_STREAM was received, the flow controller was already
 		// informed about the final byteOffset for this stream
 		if !s.cancelledRemotely {
-			if queueMaxStreamData := s.flowController.AddBytesRead(protocol.ByteCount(m)); queueMaxStreamData {
+			hasStream, hasConn := s.flowController.AddBytesRead(protocol.ByteCount(m))
+			if hasStream {
 				s.queuedMaxStreamData = true
-				queuedNewControlFrame = true
+				hasStreamWindowUpdate = true
+			}
+			if hasConn {
+				hasConnWindowUpdate = true
 			}
 		}
 
@@ -217,10 +223,10 @@ func (s *receiveStream) readImpl(p []byte) (bool, int, error) {
 				s.currentFrameDone()
 			}
 			s.errorRead = true
-			return queuedNewControlFrame, bytesRead, io.EOF
+			return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, io.EOF
 		}
 	}
-	return queuedNewControlFrame, bytesRead, nil
+	return hasStreamWindowUpdate, hasConnWindowUpdate, bytesRead, nil
 }
 
 func (s *receiveStream) dequeueNextFrame() {
diff --git a/stream.go b/stream.go
index b43a9702..9cd2695d 100644
--- a/stream.go
+++ b/stream.go
@@ -24,6 +24,7 @@ var errDeadline net.Error = &deadlineError{}
 
 // The streamSender is notified by the stream about various events.
 type streamSender interface {
+	onHasConnectionData()
 	onHasStreamData(protocol.StreamID, sendStreamI)
 	onHasStreamControlFrame(protocol.StreamID, streamControlFrameGetter)
 	// must be called without holding the mutex that is acquired by closeForShutdown


Test patch:
diff --git a/closed_conn_test.go b/closed_conn_test.go
index 4bb59e90..20a61fce 100644
--- a/closed_conn_test.go
+++ b/closed_conn_test.go
@@ -2,25 +2,33 @@ package quic
 
 import (
 	"net"
+	"testing"
 
 	"github.com/quic-go/quic-go/internal/utils"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	"github.com/stretchr/testify/require"
 )
 
-var _ = Describe("Closed local connection", func() {
-	It("repeats the packet containing the CONNECTION_CLOSE frame", func() {
-		written := make(chan net.Addr, 1)
-		conn := newClosedLocalConn(func(addr net.Addr, _ packetInfo) { written <- addr }, utils.DefaultLogger)
-		addr := &net.UDPAddr{IP: net.IPv4(127, 1, 2, 3), Port: 1337}
-		for i := 1; i <= 20; i++ {
-			conn.handlePacket(receivedPacket{remoteAddr: addr})
-			if i == 1 || i == 2 || i == 4 || i == 8 || i == 16 {
-				Expect(written).To(Receive(Equal(addr))) // receive the CONNECTION_CLOSE
-			} else {
-				Expect(written).ToNot(Receive())
+func TestClosedLocalConnection(t *testing.T) {
+	written := make(chan net.Addr, 1)
+	conn := newClosedLocalConn(func(addr net.Addr, _ packetInfo) { written <- addr }, utils.DefaultLogger)
+	addr := &net.UDPAddr{IP: net.IPv4(127, 1, 2, 3), Port: 1337}
+	for i := 1; i <= 20; i++ {
+		conn.handlePacket(receivedPacket{remoteAddr: addr})
+		if i == 1 || i == 2 || i == 4 || i == 8 || i == 16 {
+			select {
+			case gotAddr := <-written:
+				require.Equal(t, addr, gotAddr) // receive the CONNECTION_CLOSE
+			default:
+				t.Fatal("expected to receive address")
+			}
+		} else {
+			select {
+			case gotAddr := <-written:
+				t.Fatalf("unexpected address received: %v", gotAddr)
+			default:
+				// Nothing received, which is expected
 			}
 		}
-	})
-})
+	}
+}
diff --git a/crypto_stream_manager_test.go b/crypto_stream_manager_test.go
index 1f09c13d..c71bab56 100644
--- a/crypto_stream_manager_test.go
+++ b/crypto_stream_manager_test.go
@@ -1,78 +1,87 @@
 package quic
 
 import (
+	"testing"
+
 	"github.com/quic-go/quic-go/internal/protocol"
 	"github.com/quic-go/quic-go/internal/wire"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	"github.com/stretchr/testify/require"
 )
 
-var _ = Describe("Crypto Stream Manager", func() {
-	var (
-		csm *cryptoStreamManager
-
-		initialStream   *cryptoStream
-		handshakeStream *cryptoStream
-		oneRTTStream    *cryptoStream
-	)
-
-	BeforeEach(func() {
-		initialStream = newCryptoStream()
-		handshakeStream = newCryptoStream()
-		oneRTTStream = newCryptoStream()
-		csm = newCryptoStreamManager(initialStream, handshakeStream, oneRTTStream)
+func TestCryptoStreamManager(t *testing.T) {
+	t.Run("Initial", func(t *testing.T) {
+		testCryptoStreamManager(t, protocol.EncryptionInitial)
 	})
-
-	It("passes messages to the initial stream", func() {
-		cf := &wire.CryptoFrame{Data: []byte("foobar")}
-		Expect(csm.HandleCryptoFrame(cf, protocol.EncryptionInitial)).To(Succeed())
-		Expect(csm.GetCryptoData(protocol.EncryptionInitial)).To(Equal([]byte("foobar")))
+	t.Run("Handshake", func(t *testing.T) {
+		testCryptoStreamManager(t, protocol.EncryptionHandshake)
 	})
-
-	It("passes messages to the handshake stream", func() {
-		cf := &wire.CryptoFrame{Data: []byte("foobar")}
-		Expect(csm.HandleCryptoFrame(cf, protocol.EncryptionHandshake)).To(Succeed())
-		Expect(csm.GetCryptoData(protocol.EncryptionHandshake)).To(Equal([]byte("foobar")))
+	t.Run("1-RTT", func(t *testing.T) {
+		testCryptoStreamManager(t, protocol.Encryption1RTT)
 	})
+}
 
-	It("passes messages to the 1-RTT stream", func() {
-		cf := &wire.CryptoFrame{Data: []byte("foobar")}
-		Expect(csm.HandleCryptoFrame(cf, protocol.Encryption1RTT)).To(Succeed())
-		Expect(csm.GetCryptoData(protocol.Encryption1RTT)).To(Equal([]byte("foobar")))
-	})
+func testCryptoStreamManager(t *testing.T, encLevel protocol.EncryptionLevel) {
+	initialStream := newCryptoStream()
+	handshakeStream := newCryptoStream()
+	oneRTTStream := newCryptoStream()
+	csm := newCryptoStreamManager(initialStream, handshakeStream, oneRTTStream)
 
-	It("processes all messages", func() {
-		Expect(csm.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("bar"), Offset: 3}, protocol.EncryptionHandshake)).To(Succeed())
-		Expect(csm.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("foo")}, protocol.EncryptionHandshake)).To(Succeed())
-		var data []byte
-		for {
-			b := csm.GetCryptoData(protocol.EncryptionHandshake)
-			if len(b) == 0 {
-				break
-			}
-			data = append(data, b...)
+	require.NoError(t, csm.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("foo")}, encLevel))
+	require.NoError(t, csm.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("bar"), Offset: 3}, encLevel))
+	var data []byte
+	for {
+		b := csm.GetCryptoData(encLevel)
+		if len(b) == 0 {
+			break
 		}
-		Expect(data).To(Equal([]byte("foobar")))
-	})
+		data = append(data, b...)
+	}
+	require.Equal(t, []byte("foobar"), data)
+}
 
-	It("errors for unknown encryption levels", func() {
-		err := csm.HandleCryptoFrame(&wire.CryptoFrame{}, 42)
-		Expect(err).To(HaveOccurred())
-		Expect(err.Error()).To(ContainSubstring("received CRYPTO frame with unexpected encryption level"))
-	})
+func TestCryptoStreamManagerInvalidEncryptionLevel(t *testing.T) {
+	csm := newCryptoStreamManager(nil, nil, nil)
+	require.ErrorContains(t,
+		csm.HandleCryptoFrame(&wire.CryptoFrame{}, protocol.Encryption0RTT),
+		"received CRYPTO frame with unexpected encryption level",
+	)
+}
 
-	It("drops Initial", func() {
-		Expect(initialStream.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("foo")})).To(Succeed())
-		err := csm.Drop(protocol.EncryptionInitial)
-		Expect(err).To(HaveOccurred())
-		Expect(err.Error()).To(ContainSubstring("encryption level changed, but crypto stream has more data to read"))
+func TestCryptoStreamManagerDropEncryptionLevel(t *testing.T) {
+	t.Run("Initial", func(t *testing.T) {
+		testCryptoStreamManagerDropEncryptionLevel(t, protocol.EncryptionInitial)
 	})
-
-	It("drops Handshake", func() {
-		Expect(handshakeStream.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("foo")})).To(Succeed())
-		err := csm.Drop(protocol.EncryptionHandshake)
-		Expect(err).To(HaveOccurred())
-		Expect(err.Error()).To(ContainSubstring("encryption level changed, but crypto stream has more data to read"))
+	t.Run("Handshake", func(t *testing.T) {
+		testCryptoStreamManagerDropEncryptionLevel(t, protocol.EncryptionHandshake)
 	})
-})
+}
+
+func testCryptoStreamManagerDropEncryptionLevel(t *testing.T, encLevel protocol.EncryptionLevel) {
+	initialStream := newCryptoStream()
+	handshakeStream := newCryptoStream()
+	oneRTTStream := newCryptoStream()
+	csm := newCryptoStreamManager(initialStream, handshakeStream, oneRTTStream)
+
+	require.NoError(t, csm.HandleCryptoFrame(&wire.CryptoFrame{Data: []byte("foo")}, encLevel))
+	require.ErrorContains(t, csm.Drop(encLevel), "encryption level changed, but crypto stream has more data to read")
+
+	require.Equal(t, []byte("foo"), csm.GetCryptoData(encLevel))
+	require.NoError(t, csm.Drop(encLevel))
+}
+
+func TestCryptoStreamManagerPostHandshake(t *testing.T) {
+	initialStream := newCryptoStream()
+	handshakeStream := newCryptoStream()
+	oneRTTStream := newCryptoStream()
+	csm := newCryptoStreamManager(initialStream, handshakeStream, oneRTTStream)
+
+	_, err := oneRTTStream.Write([]byte("foo"))
+	require.NoError(t, err)
+	_, err = oneRTTStream.Write([]byte("bar"))
+	require.NoError(t, err)
+	require.Equal(t,
+		&wire.CryptoFrame{Data: []byte("foobar")},
+		csm.GetPostHandshakeData(protocol.ByteCount(10)),
+	)
+}
diff --git a/datagram_queue_test.go b/datagram_queue_test.go
index fe8db59f..dc621183 100644
--- a/datagram_queue_test.go
+++ b/datagram_queue_test.go
@@ -3,143 +3,159 @@ package quic
 import (
 	"context"
 	"errors"
+	"testing"
 	"time"
 
 	"github.com/quic-go/quic-go/internal/utils"
 	"github.com/quic-go/quic-go/internal/wire"
 
-	. "github.com/onsi/ginkgo/v2"
-	. "github.com/onsi/gomega"
+	"github.com/stretchr/testify/require"
 )
 
-var _ = Describe("Datagram Queue", func() {
-	var queue *datagramQueue
-	var queued chan struct{}
-
-	BeforeEach(func() {
-		queued = make(chan struct{}, 100)
-		queue = newDatagramQueue(func() { queued <- struct{}{} }, utils.DefaultLogger)
-	})
-
-	Context("sending", func() {
-		It("returns nil when there's no datagram to send", func() {
-			Expect(queue.Peek()).To(BeNil())
-		})
-
-		It("queues a datagram", func() {
-			frame := &wire.DatagramFrame{Data: []byte("foobar")}
-			Expect(queue.Add(frame)).To(Succeed())
-			Expect(queued).To(HaveLen(1))
-			f := queue.Peek()
-			Expect(f.Data).To(Equal([]byte("foobar")))
-			queue.Pop()
-			Expect(queue.Peek()).To(BeNil())
-		})
-
-		It("blocks when the maximum number of datagrams have been queued", func() {
-			for i := 0; i < maxDatagramSendQueueLen; i++ {
-				Expect(queue.Add(&wire.DatagramFrame{Data: []byte{0}})).To(Succeed())
-			}
-			errChan := make(chan error, 1)
-			go func() {
-				defer GinkgoRecover()
-				errChan <- queue.Add(&wire.DatagramFrame{Data: []byte("foobar")})
-			}()
-			Consistently(errChan, 50*time.Millisecond).ShouldNot(Receive())
-			Expect(queue.Peek()).ToNot(BeNil())
-			Consistently(errChan, 50*time.Millisecond).ShouldNot(Receive())
-			queue.Pop()
-			Eventually(errChan).Should(Receive(BeNil()))
-			for i := 1; i < maxDatagramSendQueueLen; i++ {
-				queue.Pop()
-			}
-			f := queue.Peek()
-			Expect(f).ToNot(BeNil())
-			Expect(f.Data).To(Equal([]byte("foobar")))
-		})
-
-		It("returns the same datagram multiple times, when Pop isn't called", func() {
-			Expect(queue.Add(&wire.DatagramFrame{Data: []byte("foo")})).To(Succeed())
-			Expect(queue.Add(&wire.DatagramFrame{Data: []byte("bar")})).To(Succeed())
-
-			Eventually(queued).Should(HaveLen(2))
-			f := queue.Peek()
-			Expect(f.Data).To(Equal([]byte("foo")))
-			Expect(queue.Peek()).To(Equal(f))
-			Expect(queue.Peek()).To(Equal(f))
-			queue.Pop()
-			f = queue.Peek()
-			Expect(f).ToNot(BeNil())
-			Expect(f.Data).To(Equal([]byte("bar")))
-		})
-
-		It("closes", func() {
-			for i := 0; i < maxDatagramSendQueueLen; i++ {
-				Expect(queue.Add(&wire.DatagramFrame{Data: []byte("foo")})).To(Succeed())
-			}
-			errChan := make(chan error, 1)
-			go func() {
-				defer GinkgoRecover()
-				errChan <- queue.Add(&wire.DatagramFrame{Data: []byte("foo")})
-			}()
-			Consistently(errChan, 25*time.Millisecond).ShouldNot(Receive())
-			testErr := errors.New("test error")
-			queue.CloseWithError(testErr)
-			Eventually(errChan).Should(Receive(MatchError(testErr)))
-		})
-	})
-
-	Context("receiving", func() {
-		It("receives DATAGRAM frames", func() {
-			queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("foo")})
-			queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("bar")})
-			data, err := queue.Receive(context.Background())
-			Expect(err).ToNot(HaveOccurred())
-			Expect(data).To(Equal([]byte("foo")))
-			data, err = queue.Receive(context.Background())
-			Expect(err).ToNot(HaveOccurred())
-			Expect(data).To(Equal([]byte("bar")))
-		})
-
-		It("blocks until a frame is received", func() {
-			c := make(chan []byte, 1)
-			go func() {
-				defer GinkgoRecover()
-				data, err := queue.Receive(context.Background())
-				Expect(err).ToNot(HaveOccurred())
-				c <- data
-			}()
-
-			Consistently(c).ShouldNot(Receive())
-			queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("foobar")})
-			Eventually(c).Should(Receive(Equal([]byte("foobar"))))
-		})
-
-		It("blocks until context is done", func() {
-			ctx, cancel := context.WithCancel(context.Background())
-			errChan := make(chan error)
-			go func() {
-				defer GinkgoRecover()
-				_, err := queue.Receive(ctx)
-				errChan <- err
-			}()
-
-			Consistently(errChan).ShouldNot(Receive())
-			cancel()
-			Eventually(errChan).Should(Receive(Equal(context.Canceled)))
-		})
-
-		It("closes", func() {
-			errChan := make(chan error, 1)
-			go func() {
-				defer GinkgoRecover()
-				_, err := queue.Receive(context.Background())
-				errChan <- err
-			}()
-
-			Consistently(errChan).ShouldNot(Receive())
-			queue.CloseWithError(errors.New("test error"))
-			Eventually(errChan).Should(Receive(MatchError("test error")))
-		})
-	})
-})
+func TestDatagramQueuePeekAndPop(t *testing.T) {
+	var queued []struct{}
+	queue := newDatagramQueue(func() { queued = append(queued, struct{}{}) }, utils.DefaultLogger)
+	require.Nil(t, queue.Peek())
+	require.Empty(t, queued)
+	require.NoError(t, queue.Add(&wire.DatagramFrame{Data: []byte("foo")}))
+	require.Len(t, queued, 1)
+	require.Equal(t, &wire.DatagramFrame{Data: []byte("foo")}, queue.Peek())
+	// calling peek again returns the same datagram
+	require.Equal(t, &wire.DatagramFrame{Data: []byte("foo")}, queue.Peek())
+	queue.Pop()
+	require.Nil(t, queue.Peek())
+}
+
+func TestDatagramQueueSendQueueLength(t *testing.T) {
+	queue := newDatagramQueue(func() {}, utils.DefaultLogger)
+
+	for i := 0; i < maxDatagramSendQueueLen; i++ {
+		require.NoError(t, queue.Add(&wire.DatagramFrame{Data: []byte{0}}))
+	}
+	errChan := make(chan error, 1)
+	go func() { errChan <- queue.Add(&wire.DatagramFrame{Data: []byte("foobar")}) }()
+
+	select {
+	case <-errChan:
+		t.Fatal("expected to not receive error")
+	case <-time.After(scaleDuration(10 * time.Millisecond)):
+	}
+
+	// peeking doesn't remove the datagram from the queue...
+	require.NotNil(t, queue.Peek())
+	select {
+	case <-errChan:
+		t.Fatal("expected to not receive error")
+	case <-time.After(scaleDuration(10 * time.Millisecond)):
+	}
+
+	// ...but popping does
+	queue.Pop()
+	select {
+	case err := <-errChan:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+	// pop all the remaining datagrams
+	for i := 1; i < maxDatagramSendQueueLen; i++ {
+		queue.Pop()
+	}
+	f := queue.Peek()
+	require.NotNil(t, f)
+	require.Equal(t, &wire.DatagramFrame{Data: []byte("foobar")}, f)
+}
+
+func TestDatagramQueueReceive(t *testing.T) {
+	queue := newDatagramQueue(func() {}, utils.DefaultLogger)
+
+	// receive frames that were received earlier
+	queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("foo")})
+	queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("bar")})
+	data, err := queue.Receive(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, []byte("foo"), data)
+	data, err = queue.Receive(context.Background())
+	require.NoError(t, err)
+	require.Equal(t, []byte("bar"), data)
+}
+
+func TestDatagramQueueReceiveBlocking(t *testing.T) {
+	queue := newDatagramQueue(func() {}, utils.DefaultLogger)
+
+	// block until a new frame is received
+	type result struct {
+		data []byte
+		err  error
+	}
+	resultChan := make(chan result, 1)
+	go func() {
+		data, err := queue.Receive(context.Background())
+		resultChan <- result{data, err}
+	}()
+
+	select {
+	case <-resultChan:
+		t.Fatal("expected to not receive result")
+	case <-time.After(scaleDuration(10 * time.Millisecond)):
+	}
+	queue.HandleDatagramFrame(&wire.DatagramFrame{Data: []byte("foobar")})
+	select {
+	case result := <-resultChan:
+		require.NoError(t, result.err)
+		require.Equal(t, []byte("foobar"), result.data)
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+
+	// unblock when the context is canceled
+	ctx, cancel := context.WithCancel(context.Background())
+	errChan := make(chan error, 1)
+	go func() {
+		_, err := queue.Receive(ctx)
+		errChan <- err
+	}()
+	select {
+	case <-errChan:
+		t.Fatal("expected to not receive error")
+	case <-time.After(scaleDuration(10 * time.Millisecond)):
+	}
+	cancel()
+	select {
+	case err := <-errChan:
+		require.ErrorIs(t, err, context.Canceled)
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+}
+
+func TestDatagramQueueClose(t *testing.T) {
+	queue := newDatagramQueue(func() {}, utils.DefaultLogger)
+
+	for i := 0; i < maxDatagramSendQueueLen; i++ {
+		require.NoError(t, queue.Add(&wire.DatagramFrame{Data: []byte{0}}))
+	}
+	errChan1 := make(chan error, 1)
+	go func() { errChan1 <- queue.Add(&wire.DatagramFrame{Data: []byte("foobar")}) }()
+	errChan2 := make(chan error, 1)
+	go func() {
+		_, err := queue.Receive(context.Background())
+		errChan2 <- err
+	}()
+
+	queue.CloseWithError(errors.New("test error"))
+
+	select {
+	case err := <-errChan1:
+		require.EqualError(t, err, "test error")
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+
+	select {
+	case err := <-errChan2:
+		require.EqualError(t, err, "test error")
+	case <-time.After(time.Second):
+		t.Fatal("timeout")
+	}
+}
diff --git a/internal/flowcontrol/connection_flow_controller_test.go b/internal/flowcontrol/connection_flow_controller_test.go
index de54ca28..74b4a5f7 100644
--- a/internal/flowcontrol/connection_flow_controller_test.go
+++ b/internal/flowcontrol/connection_flow_controller_test.go
@@ -19,8 +19,9 @@ func TestConnectionFlowControlWindowUpdate(t *testing.T) {
 		&utils.RTTStats{},
 		utils.DefaultLogger,
 	)
+	require.False(t, fc.AddBytesRead(1))
 	require.Zero(t, fc.GetWindowUpdate(time.Now()))
-	fc.AddBytesRead(100)
+	require.True(t, fc.AddBytesRead(99))
 	require.Equal(t, protocol.ByteCount(200), fc.GetWindowUpdate(time.Now()))
 }
 
diff --git a/internal/flowcontrol/stream_flow_controller_test.go b/internal/flowcontrol/stream_flow_controller_test.go
index 34bc8084..92d57266 100644
--- a/internal/flowcontrol/stream_flow_controller_test.go
+++ b/internal/flowcontrol/stream_flow_controller_test.go
@@ -192,16 +192,20 @@ func TestStreamWindowUpdate(t *testing.T) {
 		utils.DefaultLogger,
 	)
 	require.Zero(t, fc.GetWindowUpdate(time.Now()))
-	fc.AddBytesRead(24)
+	hasStreamWindowUpdate, _ := fc.AddBytesRead(24)
+	require.False(t, hasStreamWindowUpdate)
 	require.Zero(t, fc.GetWindowUpdate(time.Now()))
 	// the window is updated when it's 25% filled
-	fc.AddBytesRead(1)
+	hasStreamWindowUpdate, _ = fc.AddBytesRead(1)
+	require.True(t, hasStreamWindowUpdate)
 	require.Equal(t, protocol.ByteCount(125), fc.GetWindowUpdate(time.Now()))
 
-	fc.AddBytesRead(24)
+	hasStreamWindowUpdate, _ = fc.AddBytesRead(24)
+	require.False(t, hasStreamWindowUpdate)
 	require.Zero(t, fc.GetWindowUpdate(time.Now()))
 	// the window is updated when it's 25% filled
-	fc.AddBytesRead(1)
+	hasStreamWindowUpdate, _ = fc.AddBytesRead(1)
+	require.True(t, hasStreamWindowUpdate)
 	require.Equal(t, protocol.ByteCount(150), fc.GetWindowUpdate(time.Now()))
 
 	// Receive the final offset.
@@ -211,6 +215,31 @@ func TestStreamWindowUpdate(t *testing.T) {
 	require.Zero(t, fc.GetWindowUpdate(time.Now()))
 }
 
+func TestStreamConnectionWindowUpdate(t *testing.T) {
+	connFC := NewConnectionFlowController(
+		100,
+		protocol.MaxByteCount,
+		nil,
+		&utils.RTTStats{},
+		utils.DefaultLogger,
+	)
+	fc := NewStreamFlowController(
+		42,
+		connFC,
+		1000,
+		protocol.MaxByteCount,
+		protocol.MaxByteCount,
+		&utils.RTTStats{},
+		utils.DefaultLogger,
+	)
+
+	hasStreamWindowUpdate, hasConnWindowUpdate := fc.AddBytesRead(50)
+	require.False(t, hasStreamWindowUpdate)
+	require.Zero(t, fc.GetWindowUpdate(time.Now()))
+	require.True(t, hasConnWindowUpdate)
+	require.NotZero(t, connFC.GetWindowUpdate(time.Now()))
+}
+
 func TestStreamWindowAutoTuning(t *testing.T) {
 	// the RTT is 1 second
 	rttStats := &utils.RTTStats{}
diff --git a/mock_stream_sender_test.go b/mock_stream_sender_test.go
index d3418ee8..7d3a76e5 100644
--- a/mock_stream_sender_test.go
+++ b/mock_stream_sender_test.go
@@ -40,6 +40,42 @@ func (m *MockStreamSender) EXPECT() *MockStreamSenderMockRecorder {
 	return m.recorder
 }
 
+// onHasConnectionData mocks base method.
+func (m *MockStreamSender) onHasConnectionData() {
+	m.ctrl.T.Helper()
+	m.ctrl.Call(m, "onHasConnectionData")
+}
+
+// onHasConnectionData indicates an expected call of onHasConnectionData.
+func (mr *MockStreamSenderMockRecorder) onHasConnectionData() *MockStreamSenderonHasConnectionDataCall {
+	mr.mock.ctrl.T.Helper()
+	call := mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "onHasConnectionData", reflect.TypeOf((*MockStreamSender)(nil).onHasConnectionData))
+	return &MockStreamSenderonHasConnectionDataCall{Call: call}
+}
+
+// MockStreamSenderonHasConnectionDataCall wrap *gomock.Call
+type MockStreamSenderonHasConnectionDataCall struct {
+	*gomock.Call
+}
+
+// Return rewrite *gomock.Call.Return
+func (c *MockStreamSenderonHasConnectionDataCall) Return() *MockStreamSenderonHasConnectionDataCall {
+	c.Call = c.Call.Return()
+	return c
+}
+
+// Do rewrite *gomock.Call.Do
+func (c *MockStreamSenderonHasConnectionDataCall) Do(f func()) *MockStreamSenderonHasConnectionDataCall {
+	c.Call = c.Call.Do(f)
+	return c
+}
+
+// DoAndReturn rewrite *gomock.Call.DoAndReturn
+func (c *MockStreamSenderonHasConnectionDataCall) DoAndReturn(f func()) *MockStreamSenderonHasConnectionDataCall {
+	c.Call = c.Call.DoAndReturn(f)
+	return c
+}
+
 // onHasStreamControlFrame mocks base method.
 func (m *MockStreamSender) onHasStreamControlFrame(arg0 protocol.StreamID, arg1 streamControlFrameGetter) {
 	m.ctrl.T.Helper()
diff --git a/receive_stream_test.go b/receive_stream_test.go
index b4ce411d..a2ef35ec 100644
--- a/receive_stream_test.go
+++ b/receive_stream_test.go
@@ -152,7 +152,17 @@ func TestReceiveStreamReadOverlappingData(t *testing.T) {
 	require.Equal(t, []byte{'f', 'o', 'o', 'b', 'a', 'r'}, b)
 }
 
-func TestReceiveStreamMaxStreamData(t *testing.T) {
+func TestReceiveStreamFlowControlUpdates(t *testing.T) {
+	t.Run("stream", func(t *testing.T) {
+		testReceiveStreamFlowControlUpdates(t, true, false)
+	})
+
+	t.Run("connection", func(t *testing.T) {
+		testReceiveStreamFlowControlUpdates(t, false, true)
+	})
+}
+
+func testReceiveStreamFlowControlUpdates(t *testing.T, hasStreamWindowUpdate, hasConnWindowUpdate bool) {
 	const streamID protocol.StreamID = 42
 	mockCtrl := gomock.NewController(t)
 	mockFC := mocks.NewMockStreamFlowController(mockCtrl)
@@ -163,19 +173,31 @@ func TestReceiveStreamMaxStreamData(t *testing.T) {
 	mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(4), false, now)
 	require.NoError(t, str.handleStreamFrame(&wire.StreamFrame{Data: []byte{0xde, 0xad, 0xbe, 0xef}}, now))
 
-	mockFC.EXPECT().AddBytesRead(protocol.ByteCount(3)).Return(true)
-	mockSender.EXPECT().onHasStreamControlFrame(streamID, str)
+	mockFC.EXPECT().AddBytesRead(protocol.ByteCount(3)).Return(hasStreamWindowUpdate, hasConnWindowUpdate)
+	if hasStreamWindowUpdate {
+		mockSender.EXPECT().onHasStreamControlFrame(streamID, str)
+	}
+	if hasConnWindowUpdate {
+		mockSender.EXPECT().onHasConnectionData()
+	}
 	n, err := (&readerWithTimeout{Reader: str, Timeout: time.Second}).Read(make([]byte, 3))
 	require.NoError(t, err)
 	require.Equal(t, 3, n)
 	require.True(t, mockCtrl.Satisfied())
 
-	now = now.Add(time.Second)
-	mockFC.EXPECT().GetWindowUpdate(now).Return(protocol.ByteCount(1337))
-	f, ok, hasMore := str.getControlFrame(now)
-	require.True(t, ok)
-	require.Equal(t, &wire.MaxStreamDataFrame{StreamID: streamID, MaximumStreamData: 1337}, f.Frame)
-	require.False(t, hasMore)
+	if hasStreamWindowUpdate {
+		now = now.Add(time.Second)
+		mockFC.EXPECT().GetWindowUpdate(now).Return(protocol.ByteCount(1337))
+		f, ok, hasMore := str.getControlFrame(now)
+		require.True(t, ok)
+		require.Equal(t, &wire.MaxStreamDataFrame{StreamID: streamID, MaximumStreamData: 1337}, f.Frame)
+		require.False(t, hasMore)
+	}
+	if hasConnWindowUpdate {
+		_, ok, hasMore := str.getControlFrame(now)
+		require.False(t, ok)
+		require.False(t, hasMore)
+	}
 }
 
 func TestReceiveStreamDeadlineInThePast(t *testing.T) {
@@ -583,9 +605,9 @@ func TestReceiveStreamConcurrentReads(t *testing.T) {
 
 	mockFC.EXPECT().UpdateHighestReceived(protocol.ByteCount(6), gomock.Any(), gomock.Any()).AnyTimes()
 	var bytesRead protocol.ByteCount
-	mockFC.EXPECT().AddBytesRead(gomock.Any()).Do(func(n protocol.ByteCount) bool {
+	mockFC.EXPECT().AddBytesRead(gomock.Any()).Do(func(n protocol.ByteCount) (bool, bool) {
 		bytesRead += n
-		return false
+		return false, false
 	}).AnyTimes()
 
 	var numCompleted atomic.Int32