diff --git a/internal/quic/config.go b/internal/quic/config.go
index d68e2c7..b390d69 100644
--- a/internal/quic/config.go
+++ b/internal/quic/config.go
@@ -41,6 +41,12 @@
 	// If zero, the default value of 1MiB is used.
 	// If negative, the limit is zero.
 	MaxStreamWriteBufferSize int64
+
+	// MaxConnReadBufferSize is the maximum amount of data sent by the peer that a
+	// connection will buffer for reading, across all streams.
+	// If zero, the default value of 1MiB is used.
+	// If negative, the limit is zero.
+	MaxConnReadBufferSize int64
 }
 
 func configDefault(v, def, limit int64) int64 {
@@ -69,3 +75,7 @@
 func (c *Config) maxStreamWriteBufferSize() int64 {
 	return configDefault(c.MaxStreamWriteBufferSize, 1<<20, maxVarint)
 }
+
+func (c *Config) maxConnReadBufferSize() int64 {
+	return configDefault(c.MaxConnReadBufferSize, 1<<20, maxVarint)
+}
diff --git a/internal/quic/config_test.go b/internal/quic/config_test.go
index b99ffef..d292854 100644
--- a/internal/quic/config_test.go
+++ b/internal/quic/config_test.go
@@ -10,6 +10,7 @@
 
 func TestConfigTransportParameters(t *testing.T) {
 	const (
+		wantInitialMaxData        = int64(1)
 		wantInitialMaxStreamData  = int64(2)
 		wantInitialMaxStreamsBidi = int64(3)
 		wantInitialMaxStreamsUni  = int64(4)
@@ -18,12 +19,16 @@
 		c.MaxBidiRemoteStreams = wantInitialMaxStreamsBidi
 		c.MaxUniRemoteStreams = wantInitialMaxStreamsUni
 		c.MaxStreamReadBufferSize = wantInitialMaxStreamData
+		c.MaxConnReadBufferSize = wantInitialMaxData
 	})
 	tc.handshake()
 	if tc.sentTransportParameters == nil {
 		t.Fatalf("conn didn't send transport parameters during handshake")
 	}
 	p := tc.sentTransportParameters
+	if got, want := p.initialMaxData, wantInitialMaxData; got != want {
+		t.Errorf("initial_max_data = %v, want %v", got, want)
+	}
 	if got, want := p.initialMaxStreamDataBidiLocal, wantInitialMaxStreamData; got != want {
 		t.Errorf("initial_max_stream_data_bidi_local = %v, want %v", got, want)
 	}
diff --git a/internal/quic/conn.go b/internal/quic/conn.go
index 117364f..0ab6f69 100644
--- a/internal/quic/conn.go
+++ b/internal/quic/conn.go
@@ -117,6 +117,7 @@
 		maxUDPPayloadSize:              maxUDPPayloadSize,
 		maxAckDelay:                    maxAckDelay,
 		disableActiveMigration:         true,
+		initialMaxData:                 config.maxConnReadBufferSize(),
 		initialMaxStreamDataBidiLocal:  config.maxStreamReadBufferSize(),
 		initialMaxStreamDataBidiRemote: config.maxStreamReadBufferSize(),
 		initialMaxStreamDataUni:        config.maxStreamReadBufferSize(),
diff --git a/internal/quic/conn_flow.go b/internal/quic/conn_flow.go
new file mode 100644
index 0000000..790210b
--- /dev/null
+++ b/internal/quic/conn_flow.go
@@ -0,0 +1,111 @@
+// 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 (
+	"sync/atomic"
+	"time"
+)
+
+// connInflow tracks connection-level flow control for data sent by the peer to us.
+//
+// There are four byte offsets of significance in the stream of data received from the peer,
+// each >= to the previous:
+//
+//   - bytes read by the user
+//   - bytes received from the peer
+//   - limit sent to the peer in a MAX_DATA frame
+//   - potential new limit to sent to the peer
+//
+// We maintain a flow control window, so as bytes are read by the user
+// the potential limit is extended correspondingly.
+//
+// We keep an atomic counter of bytes read by the user and not yet applied to the
+// potential limit (credit). When this count grows large enough, we update the
+// new limit to send and mark that we need to send a new MAX_DATA frame.
+type connInflow struct {
+	sent      sentVal // set when we need to send a MAX_DATA update to the peer
+	usedLimit int64   // total bytes sent by the peer, must be less than sentLimit
+	sentLimit int64   // last MAX_DATA sent to the peer
+	newLimit  int64   // new MAX_DATA to send
+
+	credit atomic.Int64 // bytes read but not yet applied to extending the flow-control window
+}
+
+func (c *Conn) inflowInit() {
+	// The initial MAX_DATA limit is sent as a transport parameter.
+	c.streams.inflow.sentLimit = c.config.maxConnReadBufferSize()
+	c.streams.inflow.newLimit = c.streams.inflow.sentLimit
+}
+
+// handleStreamBytesReadOffLoop records that the user has consumed bytes from a stream.
+// We may extend the peer's flow control window.
+//
+// This is called indirectly by the user, via Read or CloseRead.
+func (c *Conn) handleStreamBytesReadOffLoop(n int64) {
+	if c.shouldUpdateFlowControl(c.streams.inflow.credit.Add(n)) {
+		// We should send a MAX_DATA update to the peer.
+		// Record this on the Conn's main loop.
+		c.sendMsg(func(now time.Time, c *Conn) {
+			c.sendMaxDataUpdate()
+		})
+	}
+}
+
+// handleStreamBytesReadOnLoop extends the peer's flow control window after
+// data has been discarded due to a RESET_STREAM frame.
+//
+// This is called on the conn's loop.
+func (c *Conn) handleStreamBytesReadOnLoop(n int64) {
+	if c.shouldUpdateFlowControl(c.streams.inflow.credit.Add(n)) {
+		c.sendMaxDataUpdate()
+	}
+}
+
+func (c *Conn) sendMaxDataUpdate() {
+	c.streams.inflow.sent.setUnsent()
+	// Apply current credit to the limit.
+	// We don't strictly need to do this here
+	// since appendMaxDataFrame will do so as well,
+	// but this avoids redundant trips down this path
+	// if the MAX_DATA frame doesn't go out right away.
+	c.streams.inflow.newLimit += c.streams.inflow.credit.Swap(0)
+}
+
+func (c *Conn) shouldUpdateFlowControl(credit int64) bool {
+	return shouldUpdateFlowControl(c.config.maxConnReadBufferSize(), credit)
+}
+
+// handleStreamBytesReceived records that the peer has sent us stream data.
+func (c *Conn) handleStreamBytesReceived(n int64) error {
+	c.streams.inflow.usedLimit += n
+	if c.streams.inflow.usedLimit > c.streams.inflow.sentLimit {
+		return localTransportError(errFlowControl)
+	}
+	return nil
+}
+
+// appendMaxDataFrame appends a MAX_DATA frame to the current packet.
+//
+// It returns true if no more frames need appending,
+// false if it could not fit a frame in the current packet.
+func (c *Conn) appendMaxDataFrame(w *packetWriter, pnum packetNumber, pto bool) bool {
+	if c.streams.inflow.sent.shouldSendPTO(pto) {
+		// Add any unapplied credit to the new limit now.
+		c.streams.inflow.newLimit += c.streams.inflow.credit.Swap(0)
+		if !w.appendMaxDataFrame(c.streams.inflow.newLimit) {
+			return false
+		}
+		c.streams.inflow.sent.setSent(pnum)
+	}
+	return true
+}
+
+// ackOrLossMaxData records the fate of a MAX_DATA frame.
+func (c *Conn) ackOrLossMaxData(pnum packetNumber, fate packetFate) {
+	c.streams.inflow.sent.ackLatestOrLoss(pnum, fate)
+}
diff --git a/internal/quic/conn_flow_test.go b/internal/quic/conn_flow_test.go
new file mode 100644
index 0000000..f01a738
--- /dev/null
+++ b/internal/quic/conn_flow_test.go
@@ -0,0 +1,186 @@
+// 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 "testing"
+
+func TestConnInflowReturnOnRead(t *testing.T) {
+	ctx := canceledContext()
+	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+		c.MaxConnReadBufferSize = 64
+	})
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   s.id,
+		data: make([]byte, 64),
+	})
+	const readSize = 8
+	if n, err := s.ReadContext(ctx, make([]byte, readSize)); n != readSize || err != nil {
+		t.Fatalf("s.Read() = %v, %v; want %v, nil", n, err, readSize)
+	}
+	tc.wantFrame("available window increases, send a MAX_DATA",
+		packetType1RTT, debugFrameMaxData{
+			max: 64 + readSize,
+		})
+	if n, err := s.ReadContext(ctx, make([]byte, 64)); n != 64-readSize || err != nil {
+		t.Fatalf("s.Read() = %v, %v; want %v, nil", n, err, 64-readSize)
+	}
+	tc.wantFrame("available window increases, send a MAX_DATA",
+		packetType1RTT, debugFrameMaxData{
+			max: 128,
+		})
+}
+
+func TestConnInflowReturnOnClose(t *testing.T) {
+	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+		c.MaxConnReadBufferSize = 64
+	})
+	tc.ignoreFrame(frameTypeStopSending)
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   s.id,
+		data: make([]byte, 64),
+	})
+	s.CloseRead()
+	tc.wantFrame("closing stream updates connection-level flow control",
+		packetType1RTT, debugFrameMaxData{
+			max: 128,
+		})
+}
+
+func TestConnInflowReturnOnReset(t *testing.T) {
+	tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+		c.MaxConnReadBufferSize = 64
+	})
+	tc.ignoreFrame(frameTypeStopSending)
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   s.id,
+		data: make([]byte, 32),
+	})
+	tc.writeFrames(packetType1RTT, debugFrameResetStream{
+		id:        s.id,
+		finalSize: 64,
+	})
+	s.CloseRead()
+	tc.wantFrame("receiving stream reseet updates connection-level flow control",
+		packetType1RTT, debugFrameMaxData{
+			max: 128,
+		})
+}
+
+func TestConnInflowStreamViolation(t *testing.T) {
+	tc := newTestConn(t, serverSide, func(c *Config) {
+		c.MaxConnReadBufferSize = 100
+	})
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+	// Total MAX_DATA consumed: 50
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, bidiStream, 0),
+		data: make([]byte, 50),
+	})
+	// Total MAX_DATA consumed: 80
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, uniStream, 0),
+		off:  20,
+		data: make([]byte, 10),
+	})
+	// Total MAX_DATA consumed: 100
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:  newStreamID(clientSide, bidiStream, 0),
+		off: 70,
+		fin: true,
+	})
+	// This stream has already consumed quota for these bytes.
+	// Total MAX_DATA consumed: 100
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, uniStream, 0),
+		data: make([]byte, 20),
+	})
+	tc.wantIdle("peer has consumed all MAX_DATA quota")
+
+	// Total MAX_DATA consumed: 101
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, bidiStream, 2),
+		data: make([]byte, 1),
+	})
+	tc.wantFrame("peer violates MAX_DATA limit",
+		packetType1RTT, debugFrameConnectionCloseTransport{
+			code: errFlowControl,
+		})
+}
+
+func TestConnInflowResetViolation(t *testing.T) {
+	tc := newTestConn(t, serverSide, func(c *Config) {
+		c.MaxConnReadBufferSize = 100
+	})
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+	tc.writeFrames(packetType1RTT, debugFrameStream{
+		id:   newStreamID(clientSide, bidiStream, 0),
+		data: make([]byte, 100),
+	})
+	tc.wantIdle("peer has consumed all MAX_DATA quota")
+
+	tc.writeFrames(packetType1RTT, debugFrameResetStream{
+		id:        newStreamID(clientSide, uniStream, 0),
+		finalSize: 0,
+	})
+	tc.wantIdle("stream reset does not consume MAX_DATA quota, no error")
+
+	tc.writeFrames(packetType1RTT, debugFrameResetStream{
+		id:        newStreamID(clientSide, uniStream, 1),
+		finalSize: 1,
+	})
+	tc.wantFrame("RESET_STREAM final size violates MAX_DATA limit",
+		packetType1RTT, debugFrameConnectionCloseTransport{
+			code: errFlowControl,
+		})
+}
+
+func TestConnInflowMultipleStreams(t *testing.T) {
+	ctx := canceledContext()
+	tc := newTestConn(t, serverSide, func(c *Config) {
+		c.MaxConnReadBufferSize = 128
+	})
+	tc.handshake()
+	tc.ignoreFrame(frameTypeAck)
+
+	var streams []*Stream
+	for _, id := range []streamID{
+		newStreamID(clientSide, uniStream, 0),
+		newStreamID(clientSide, uniStream, 1),
+		newStreamID(clientSide, bidiStream, 0),
+		newStreamID(clientSide, bidiStream, 1),
+	} {
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   id,
+			data: make([]byte, 32),
+		})
+		s, err := tc.conn.AcceptStream(ctx)
+		if err != nil {
+			t.Fatalf("AcceptStream() = %v", err)
+		}
+		streams = append(streams, s)
+		if n, err := s.ReadContext(ctx, make([]byte, 1)); err != nil || n != 1 {
+			t.Fatalf("s.Read() = %v, %v; want 1, nil", n, err)
+		}
+	}
+	tc.wantIdle("streams have read data, but not enough to update MAX_DATA")
+
+	if n, err := streams[0].ReadContext(ctx, make([]byte, 32)); err != nil || n != 31 {
+		t.Fatalf("s.Read() = %v, %v; want 31, nil", n, err)
+	}
+	tc.wantFrame("read enough data to trigger a MAX_DATA update",
+		packetType1RTT, debugFrameMaxData{
+			max: 128 + 32 + 1 + 1 + 1,
+		})
+
+	streams[2].CloseRead()
+	tc.wantFrame("closed stream triggers another MAX_DATA update",
+		packetType1RTT, debugFrameMaxData{
+			max: 128 + 32 + 1 + 32 + 1,
+		})
+}
diff --git a/internal/quic/conn_loss.go b/internal/quic/conn_loss.go
index b8146a4..85bda31 100644
--- a/internal/quic/conn_loss.go
+++ b/internal/quic/conn_loss.go
@@ -44,6 +44,8 @@
 		case frameTypeCrypto:
 			start, end := sent.nextRange()
 			c.crypto[space].ackOrLoss(start, end, fate)
+		case frameTypeMaxData:
+			c.ackOrLossMaxData(sent.num, fate)
 		case frameTypeResetStream,
 			frameTypeStopSending,
 			frameTypeMaxStreamData,
diff --git a/internal/quic/conn_loss_test.go b/internal/quic/conn_loss_test.go
index f74ec7e..9b88462 100644
--- a/internal/quic/conn_loss_test.go
+++ b/internal/quic/conn_loss_test.go
@@ -289,18 +289,58 @@
 	tc.wantIdle("no more frames sent after packet loss")
 }
 
+func TestLostMaxDataFrame(t *testing.T) {
+	// "An updated value is sent in a MAX_DATA frame if the packet
+	// containing the most recently sent MAX_DATA frame is declared lost [...]"
+	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.7
+	lostFrameTest(t, func(t *testing.T, pto bool) {
+		const maxWindowSize = 32
+		buf := make([]byte, maxWindowSize)
+		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
+			c.MaxConnReadBufferSize = 32
+		})
+
+		// We send MAX_DATA = 63.
+		tc.writeFrames(packetType1RTT, debugFrameStream{
+			id:   s.id,
+			off:  0,
+			data: make([]byte, maxWindowSize),
+		})
+		if n, err := s.Read(buf[:maxWindowSize-1]); err != nil || n != maxWindowSize-1 {
+			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, maxWindowSize-1)
+		}
+		tc.wantFrame("conn window is extended after reading data",
+			packetType1RTT, debugFrameMaxData{
+				max: (maxWindowSize * 2) - 1,
+			})
+
+		// MAX_DATA = 64, which is only one more byte, so we don't send the frame.
+		if n, err := s.Read(buf); err != nil || n != 1 {
+			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, 1)
+		}
+		tc.wantIdle("read doesn't extend window enough to send another MAX_DATA")
+
+		// The MAX_DATA = 63 packet was lost, so we send 64.
+		tc.triggerLossOrPTO(packetType1RTT, pto)
+		tc.wantFrame("resent MAX_DATA includes most current value",
+			packetType1RTT, debugFrameMaxData{
+				max: maxWindowSize * 2,
+			})
+	})
+}
+
 func TestLostMaxStreamDataFrame(t *testing.T) {
 	// "[...] an updated value is sent when the packet containing
 	// the most recent MAX_STREAM_DATA frame for a stream is lost"
 	// https://www.rfc-editor.org/rfc/rfc9000#section-13.3-3.8
 	lostFrameTest(t, func(t *testing.T, pto bool) {
-		const maxWindowSize = 10
+		const maxWindowSize = 32
 		buf := make([]byte, maxWindowSize)
 		tc, s := newTestConnAndRemoteStream(t, serverSide, uniStream, func(c *Config) {
 			c.MaxStreamReadBufferSize = maxWindowSize
 		})
 
-		// We send MAX_STREAM_DATA = 19.
+		// We send MAX_STREAM_DATA = 63.
 		tc.writeFrames(packetType1RTT, debugFrameStream{
 			id:   s.id,
 			off:  0,
@@ -315,13 +355,13 @@
 				max: (maxWindowSize * 2) - 1,
 			})
 
-		// MAX_STREAM_DATA = 20, which is only one more byte, so we don't send the frame.
+		// MAX_STREAM_DATA = 64, which is only one more byte, so we don't send the frame.
 		if n, err := s.Read(buf); err != nil || n != 1 {
 			t.Fatalf("Read() = %v, %v; want %v, nil", n, err, 1)
 		}
 		tc.wantIdle("read doesn't extend window enough to send another MAX_STREAM_DATA")
 
-		// The MAX_STREAM_DATA = 19 packet was lost, so we send 20.
+		// The MAX_STREAM_DATA = 63 packet was lost, so we send 64.
 		tc.triggerLossOrPTO(packetType1RTT, pto)
 		tc.wantFrame("resent MAX_STREAM_DATA includes most current value",
 			packetType1RTT, debugFrameMaxStreamData{
diff --git a/internal/quic/conn_streams.go b/internal/quic/conn_streams.go
index 76e9bf9..0a72d26 100644
--- a/internal/quic/conn_streams.go
+++ b/internal/quic/conn_streams.go
@@ -27,6 +27,9 @@
 	peerInitialMaxStreamDataRemote    [streamTypeCount]int64 // streams opened by us
 	peerInitialMaxStreamDataBidiLocal int64                  // streams opened by them
 
+	// Connection-level flow control.
+	inflow connInflow
+
 	// Streams with frames to send are stored in a circular linked list.
 	// sendHead is the next stream to write, or nil if there are no streams
 	// with data to send. sendTail is the last stream to write.
@@ -43,6 +46,7 @@
 	c.streams.localLimit[uniStream].init()
 	c.streams.remoteLimit[bidiStream].init(c.config.maxBidiRemoteStreams())
 	c.streams.remoteLimit[uniStream].init(c.config.maxUniRemoteStreams())
+	c.inflowInit()
 }
 
 // AcceptStream waits for and returns the next stream created by the peer.
@@ -212,6 +216,11 @@
 // It returns true if no more frames need appending,
 // false if not everything fit in the current packet.
 func (c *Conn) appendStreamFrames(w *packetWriter, pnum packetNumber, pto bool) bool {
+	// MAX_DATA
+	if !c.appendMaxDataFrame(w, pnum, pto) {
+		return false
+	}
+
 	// MAX_STREAM_DATA
 	if !c.streams.remoteLimit[uniStream].appendFrame(w, uniStream, pnum, pto) {
 		return false
diff --git a/internal/quic/stream.go b/internal/quic/stream.go
index fbc3633..84c437d 100644
--- a/internal/quic/stream.go
+++ b/internal/quic/stream.go
@@ -156,9 +156,10 @@
 	start := s.in.start
 	end := start + int64(len(b))
 	s.in.copy(start, b)
+	s.conn.handleStreamBytesReadOffLoop(int64(len(b)))
 	s.in.discardBefore(end)
 	if s.insize == -1 || s.insize > s.inwin {
-		if shouldUpdateFlowControl(s.inwin-s.in.start, s.inmaxbuf) {
+		if shouldUpdateFlowControl(s.inmaxbuf, s.in.start+s.inmaxbuf-s.inwin) {
 			// Update stream flow control with a STREAM_MAX_DATA frame.
 			s.insendmax.setUnsent()
 		}
@@ -173,10 +174,8 @@
 //
 // We want to balance keeping the peer well-supplied with flow control with not sending
 // many small updates.
-func shouldUpdateFlowControl(curwin, maxwin int64) bool {
-	// Update flow control if doing so gives the peer at least 64k tokens,
-	// or if it will double the current window.
-	return maxwin-curwin >= 64<<10 || curwin*2 < maxwin
+func shouldUpdateFlowControl(maxWindow, addedWindow int64) bool {
+	return addedWindow >= maxWindow/8
 }
 
 // Write writes data to the stream.
@@ -295,6 +294,7 @@
 	} else {
 		s.inclosed.set()
 	}
+	s.conn.handleStreamBytesReadOffLoop(s.in.end - s.in.start)
 	s.in.discardBefore(s.in.end)
 }
 
@@ -470,6 +470,12 @@
 		// Either way, we can discard this frame.
 		return nil
 	}
+	if s.insize == -1 && end > s.in.end {
+		added := end - s.in.end
+		if err := s.conn.handleStreamBytesReceived(added); err != nil {
+			return err
+		}
+	}
 	s.in.writeAt(b, off)
 	s.inset.add(off, end)
 	if fin {
@@ -492,6 +498,13 @@
 		// The stream was already reset.
 		return nil
 	}
+	if s.insize == -1 {
+		added := finalSize - s.in.end
+		if err := s.conn.handleStreamBytesReceived(added); err != nil {
+			return err
+		}
+	}
+	s.conn.handleStreamBytesReadOnLoop(finalSize - s.in.start)
 	s.in.discardBefore(s.in.end)
 	s.inresetcode = int64(code)
 	s.insize = finalSize