Repository: quic-go/quic-go
Pull request #4613 (2024-08-02 22:17:46)

only declare send stream completed when RESET_STREAM is acknowledged
~~Depends on #4610.~~

We should also do this with all the other frames.

codecov[bot]: ## [Codecov](https://app.codecov.io/gh/quic-go/quic-go/pull/4613?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 `92.30769%` with `2 lines` in your changes missing coverage. Please review.
> Project coverage is 84.99%. Comparing base [(`f9410cd`)](https://app.codecov.io/gh/quic-go/quic-go/commit/f9410cd35f7fe22018c1fc7feeacad09607507b5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go) to head [(`7fca4ab`)](https://app.codecov.io/gh/quic-go/quic-go/commit/7fca4ab8feb2d40e799dd14f5677d930907f6886?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go).

| [Files](https://app.codecov.io/gh/quic-go/quic-go/pull/4613?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 |
|---|---|---|
| [send\_stream.go](https://app.codecov.io/gh/quic-go/quic-go/pull/4613?src=pr&el=tree&filepath=send_stream.go&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=quic-go#diff-c2VuZF9zdHJlYW0uZ28=) | 91.67% | [1 Missing and 1 partial :warning: ](https://app.codecov.io/gh/quic-go/quic-go/pull/4613?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                        @@
##           stream-control-frame-sending    #4613      +/-   ##
================================================================
- Coverage                         85.10%   84.99%   -0.11%     
================================================================
  Files                               150      147       -3     
  Lines                             14796    14537     -259     
================================================================
- Hits                              12591    12355     -236     
+ Misses                             1694     1683      -11     
+ Partials                            511      499      -12     
```

</details>

[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/quic-go/quic-go/pull/4613?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/http3/client.go b/http3/client.go
index 23ac088f..e60acffe 100644
--- a/http3/client.go
+++ b/http3/client.go
@@ -88,6 +88,7 @@ func (c *SingleDestinationRoundTripper) init() {
 		c.EnableDatagrams,
 		protocol.PerspectiveClient,
 		c.Logger,
+		0,
 	)
 	// send the SETTINGs frame, using 0-RTT data, if possible
 	go func() {
diff --git a/http3/conn.go b/http3/conn.go
index df7fb282..0fd9412f 100644
--- a/http3/conn.go
+++ b/http3/conn.go
@@ -7,6 +7,7 @@ import (
 	"net"
 	"sync"
 	"sync/atomic"
+	"time"
 
 	"github.com/quic-go/quic-go"
 	"github.com/quic-go/quic-go/internal/protocol"
@@ -51,6 +52,9 @@ type connection struct {
 
 	settings         *Settings
 	receivedSettings chan struct{}
+
+	idleTimeout time.Duration
+	idleTimer   *time.Timer
 }
 
 func newConnection(
@@ -59,17 +63,27 @@ func newConnection(
 	enableDatagrams bool,
 	perspective protocol.Perspective,
 	logger *slog.Logger,
+	idleTimeout time.Duration,
 ) *connection {
-	return &connection{
+	c := &connection{
 		ctx:              ctx,
 		Connection:       quicConn,
 		perspective:      perspective,
 		logger:           logger,
+		idleTimeout:      idleTimeout,
 		enableDatagrams:  enableDatagrams,
 		decoder:          qpack.NewDecoder(func(hf qpack.HeaderField) {}),
 		receivedSettings: make(chan struct{}),
 		streams:          make(map[protocol.StreamID]*datagrammer),
 	}
+	if idleTimeout > 0 {
+		c.idleTimer = time.AfterFunc(idleTimeout, c.onIdleTimer)
+	}
+	return c
+}
+
+func (c *connection) onIdleTimer() {
+	c.CloseWithError(quic.ApplicationErrorCode(ErrCodeNoError), "idle timeout")
 }
 
 func (c *connection) clearStream(id quic.StreamID) {
@@ -77,6 +91,9 @@ func (c *connection) clearStream(id quic.StreamID) {
 	defer c.streamMx.Unlock()
 
 	delete(c.streams, id)
+	if c.idleTimeout > 0 && len(c.streams) == 0 {
+		c.idleTimer.Reset(c.idleTimeout)
+	}
 }
 
 func (c *connection) openRequestStream(
@@ -109,12 +126,24 @@ func (c *connection) acceptStream(ctx context.Context) (quic.Stream, *datagramme
 		strID := str.StreamID()
 		c.streamMx.Lock()
 		c.streams[strID] = datagrams
+		if c.idleTimeout > 0 {
+			if len(c.streams) == 1 {
+				c.idleTimer.Stop()
+			}
+		}
 		c.streamMx.Unlock()
 		str = newStateTrackingStream(str, c, datagrams)
 	}
 	return str, datagrams, nil
 }
 
+func (c *connection) CloseWithError(code quic.ApplicationErrorCode, msg string) error {
+	if c.idleTimer != nil {
+		c.idleTimer.Stop()
+	}
+	return c.Connection.CloseWithError(code, msg)
+}
+
 func (c *connection) HandleUnidirectionalStreams(hijack func(StreamType, quic.ConnectionTracingID, quic.ReceiveStream, error) (hijacked bool)) {
 	var (
 		rcvdControlStr      atomic.Bool
diff --git a/http3/server.go b/http3/server.go
index cf52c174..9e7cd644 100644
--- a/http3/server.go
+++ b/http3/server.go
@@ -198,6 +198,12 @@ type Server struct {
 	// In that case, the stream type will not be set.
 	UniStreamHijacker func(StreamType, quic.ConnectionTracingID, quic.ReceiveStream, error) (hijacked bool)
 
+	// IdleTimeout specifies how long until idle clients connection should be
+	// closed. Idle refers only to the HTTP/3 layer, activity at the QUIC layer
+	// like PING frames are not considered.
+	// If zero or negative, there is no timeout.
+	IdleTimeout time.Duration
+
 	// ConnContext optionally specifies a function that modifies the context used for a new connection c.
 	// The provided ctx has a ServerContextKey value.
 	ConnContext func(ctx context.Context, c quic.Connection) context.Context
@@ -479,8 +485,10 @@ func (s *Server) handleConn(conn quic.Connection) error {
 		s.EnableDatagrams,
 		protocol.PerspectiveServer,
 		s.Logger,
+		s.IdleTimeout,
 	)
 	go hconn.HandleUnidirectionalStreams(s.UniStreamHijacker)
+
 	// Process all requests immediately.
 	// It's the client's responsibility to decide which requests are eligible for 0-RTT.
 	for {
diff --git a/packet_packer.go b/packet_packer.go
index f505f16a..8b8a03d4 100644
--- a/packet_packer.go
+++ b/packet_packer.go
@@ -645,6 +645,9 @@ func (p *packetPacker) composeNextPacket(maxFrameSize protocol.ByteCount, onlyAc
 		pl.length += lengthAdded
 		// add handlers for the control frames that were added
 		for i := startLen; i < len(pl.frames); i++ {
+			if pl.frames[i].Handler != nil {
+				continue
+			}
 			switch pl.frames[i].Frame.(type) {
 			case *wire.PathChallengeFrame, *wire.PathResponseFrame:
 				// Path probing is currently not supported, therefore we don't need to set the OnAcked callback yet.
diff --git a/send_stream.go b/send_stream.go
index a7bb2cd6..bcaf2abf 100644
--- a/send_stream.go
+++ b/send_stream.go
@@ -26,7 +26,7 @@ type sendStreamI interface {
 type sendStream struct {
 	mutex sync.Mutex
 
-	numOutstandingFrames int64
+	numOutstandingFrames int64 // outstanding STREAM and RESET_STREAM frames
 	retransmissionQueue  []*wire.StreamFrame
 
 	ctx       context.Context
@@ -371,7 +371,7 @@ func (s *sendStream) isNewlyCompleted() bool {
 		return false
 	}
 	// We need to keep the stream around until all frames have been sent and acknowledged.
-	if s.numOutstandingFrames > 0 || len(s.retransmissionQueue) > 0 {
+	if s.numOutstandingFrames > 0 || len(s.retransmissionQueue) > 0 || s.queuedResetStreamFrame {
 		return false
 	}
 	// The stream is completed if we sent the FIN.
@@ -430,7 +430,7 @@ func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool
 			s.mutex.Unlock()
 			// The user has called CancelWrite. If the previous cancellation was
 			// because of a STOP_SENDING, we don't need to flag the error to the
-			// user any more.
+			// user anymore.
 			if completed {
 				s.sender.onStreamCompleted(s.streamID)
 			}
@@ -445,15 +445,11 @@ func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool
 	s.ctxCancel(s.cancelWriteErr)
 	s.numOutstandingFrames = 0
 	s.retransmissionQueue = nil
-	newlyCompleted := s.isNewlyCompleted()
 	s.queuedResetStreamFrame = true
 	s.mutex.Unlock()
 
 	s.signalWrite()
 	s.sender.onHasStreamControlFrame(s.streamID, s)
-	if newlyCompleted {
-		s.sender.onStreamCompleted(s.streamID)
-	}
 }
 
 func (s *sendStream) updateSendWindow(limit protocol.ByteCount) {
@@ -488,8 +484,14 @@ func (s *sendStream) getControlFrame() (_ ackhandler.Frame, ok, hasMore bool) {
 	}
 	// RESET_STREAM frame
 	s.queuedResetStreamFrame = false
+	s.numOutstandingFrames++
 	return ackhandler.Frame{
-		Frame: &wire.ResetStreamFrame{StreamID: s.streamID, FinalSize: s.writeOffset, ErrorCode: s.cancelWriteErr.ErrorCode},
+		Frame: &wire.ResetStreamFrame{
+			StreamID:  s.streamID,
+			FinalSize: s.writeOffset,
+			ErrorCode: s.cancelWriteErr.ErrorCode,
+		},
+		Handler: (*sendStreamResetStreamHandler)(s),
 	}, true, false
 }
 
@@ -539,10 +541,10 @@ func (s *sendStreamAckHandler) OnAcked(f wire.Frame) {
 	if s.numOutstandingFrames < 0 {
 		panic("numOutStandingFrames negative")
 	}
-	newlyCompleted := (*sendStream)(s).isNewlyCompleted()
+	completed := (*sendStream)(s).isNewlyCompleted()
 	s.mutex.Unlock()
 
-	if newlyCompleted {
+	if completed {
 		s.sender.onStreamCompleted(s.streamID)
 	}
 }
@@ -564,3 +566,28 @@ func (s *sendStreamAckHandler) OnLost(f wire.Frame) {
 
 	s.sender.onHasStreamData(s.streamID, (*sendStream)(s))
 }
+
+type sendStreamResetStreamHandler sendStream
+
+var _ ackhandler.FrameHandler = &sendStreamResetStreamHandler{}
+
+func (s *sendStreamResetStreamHandler) OnAcked(wire.Frame) {
+	s.mutex.Lock()
+	s.numOutstandingFrames--
+	if s.numOutstandingFrames < 0 {
+		panic("numOutStandingFrames negative")
+	}
+	completed := (*sendStream)(s).isNewlyCompleted()
+	s.mutex.Unlock()
+
+	if completed {
+		s.sender.onStreamCompleted(s.streamID)
+	}
+}
+
+func (s *sendStreamResetStreamHandler) OnLost(wire.Frame) {
+	s.mutex.Lock()
+	s.queuedResetStreamFrame = true
+	s.mutex.Unlock()
+	s.sender.onHasStreamControlFrame(s.streamID, (*sendStream)(s))
+}


Test patch:
diff --git a/http3/conn_test.go b/http3/conn_test.go
index 4be8b7ae..1b1ff257 100644
--- a/http3/conn_test.go
+++ b/http3/conn_test.go
@@ -29,6 +29,7 @@ var _ = Describe("Connection", func() {
 				false,
 				protocol.PerspectiveServer,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&settingsFrame{
@@ -62,6 +63,7 @@ var _ = Describe("Connection", func() {
 				false,
 				protocol.PerspectiveServer,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&settingsFrame{}).Append(b)
@@ -104,6 +106,7 @@ var _ = Describe("Connection", func() {
 					false,
 					protocol.PerspectiveClient,
 					nil,
+					0,
 				)
 				buf := bytes.NewBuffer(quicvarint.Append(nil, streamType))
 				str := mockquic.NewMockStream(mockCtrl)
@@ -133,6 +136,7 @@ var _ = Describe("Connection", func() {
 					false,
 					protocol.PerspectiveClient,
 					nil,
+					0,
 				)
 				buf := bytes.NewBuffer(quicvarint.Append(nil, streamType))
 				str1 := mockquic.NewMockStream(mockCtrl)
@@ -169,6 +173,7 @@ var _ = Describe("Connection", func() {
 				false,
 				protocol.PerspectiveServer,
 				nil,
+				0,
 			)
 			buf := bytes.NewBuffer(quicvarint.Append(nil, 0x1337))
 			str := mockquic.NewMockStream(mockCtrl)
@@ -195,6 +200,7 @@ var _ = Describe("Connection", func() {
 				false,
 				protocol.PerspectiveServer,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&dataFrame{}).Append(b)
@@ -226,6 +232,7 @@ var _ = Describe("Connection", func() {
 				false,
 				protocol.PerspectiveServer,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&settingsFrame{}).Append(b)
@@ -264,6 +271,7 @@ var _ = Describe("Connection", func() {
 					false,
 					pers.Opposite(),
 					nil,
+					0,
 				)
 				buf := bytes.NewBuffer(quicvarint.Append(nil, streamTypePushStream))
 				controlStr := mockquic.NewMockStream(mockCtrl)
@@ -294,6 +302,7 @@ var _ = Describe("Connection", func() {
 				true,
 				protocol.PerspectiveClient,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&settingsFrame{Datagram: true}).Append(b)
@@ -333,6 +342,7 @@ var _ = Describe("Connection", func() {
 				true,
 				protocol.PerspectiveClient,
 				nil,
+				0,
 			)
 			b := quicvarint.Append(nil, streamTypeControlStream)
 			b = (&settingsFrame{Datagram: true}).Append(b)
diff --git a/http3/http_stream_test.go b/http3/http_stream_test.go
index a4c889e2..59f68279 100644
--- a/http3/http_stream_test.go
+++ b/http3/http_stream_test.go
@@ -43,7 +43,7 @@ var _ = Describe("Stream", func() {
 				errorCbCalled = true
 				return nil
 			}).AnyTimes()
-			str = newStream(qstr, newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil), nil)
+			str = newStream(qstr, newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil, 0), nil)
 		})
 
 		It("reads DATA frames in a single run", func() {
@@ -171,7 +171,7 @@ var _ = Describe("Request Stream", func() {
 		requestWriter := newRequestWriter()
 		conn := mockquic.NewMockEarlyConnection(mockCtrl)
 		str = newRequestStream(
-			newStream(qstr, newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil), nil),
+			newStream(qstr, newConnection(context.Background(), conn, false, protocol.PerspectiveClient, nil, 0), nil),
 			requestWriter,
 			make(chan struct{}),
 			qpack.NewDecoder(func(qpack.HeaderField) {}),
diff --git a/http3/server_test.go b/http3/server_test.go
index 7e2138d5..f7a4c9ce 100644
--- a/http3/server_test.go
+++ b/http3/server_test.go
@@ -148,7 +148,7 @@ var _ = Describe("Server", func() {
 			qconn.EXPECT().LocalAddr().AnyTimes()
 			qconn.EXPECT().ConnectionState().Return(quic.ConnectionState{}).AnyTimes()
 			qconn.EXPECT().Context().Return(context.Background()).AnyTimes()
-			conn = newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil)
+			conn = newConnection(context.Background(), qconn, false, protocol.PerspectiveServer, nil, 0)
 		})
 
 		It("calls the HTTP handler function", func() {
diff --git a/integrationtests/self/http_test.go b/integrationtests/self/http_test.go
index e2cf5fb2..cd97896b 100644
--- a/integrationtests/self/http_test.go
+++ b/integrationtests/self/http_test.go
@@ -128,6 +128,18 @@ var _ = Describe("HTTP tests", func() {
 		client = &http.Client{Transport: rt}
 	})
 
+	It("closes the connection after idle timeout", func() {
+		server.IdleTimeout = 100 * time.Millisecond
+		_, err := client.Get(fmt.Sprintf("https://localhost:%d/hello", port))
+		Expect(err).ToNot(HaveOccurred())
+
+		time.Sleep(150 * time.Millisecond)
+
+		_, err = client.Get(fmt.Sprintf("https://localhost:%d/hello", port))
+		Expect(err).ToNot(MatchError("idle timeout"))
+		server.IdleTimeout = 0
+	})
+
 	It("downloads a hello", func() {
 		resp, err := client.Get(fmt.Sprintf("https://localhost:%d/hello", port))
 		Expect(err).ToNot(HaveOccurred())
diff --git a/packet_packer_test.go b/packet_packer_test.go
index c676b2b4..1dc2494d 100644
--- a/packet_packer_test.go
+++ b/packet_packer_test.go
@@ -279,10 +279,7 @@ var _ = Describe("Packet packer", func() {
 				pnManager.EXPECT().PopPacketNumber(protocol.Encryption0RTT).Return(protocol.PacketNumber(0x42))
 				cf := ackhandler.Frame{Frame: &wire.MaxDataFrame{MaximumData: 0x1337}}
 				framer.EXPECT().HasData().Return(true)
-				framer.EXPECT().AppendControlFrames(gomock.Any(), gomock.Any(), protocol.Version1).DoAndReturn(func(frames []ackhandler.Frame, _ protocol.ByteCount, v protocol.Version) ([]ackhandler.Frame, protocol.ByteCount) {
-					Expect(frames).To(BeEmpty())
-					return append(frames, cf), cf.Frame.Length(v)
-				})
+				expectAppendControlFrames(cf)
 				// TODO: check sizes
 				framer.EXPECT().AppendStreamFrames(gomock.Any(), gomock.Any(), protocol.Version1).DoAndReturn(func(frames []ackhandler.StreamFrame, _ protocol.ByteCount, _ protocol.Version) ([]ackhandler.StreamFrame, protocol.ByteCount) {
 					return frames, 0
@@ -515,14 +512,14 @@ var _ = Describe("Packet packer", func() {
 				Expect(p.Ack).To(Equal(ack))
 			})
 
-			It("packs control frames", func() {
+			It("packs control frames, and sets OnLost / OnAcked handlers", func() {
 				pnManager.EXPECT().PeekPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42), protocol.PacketNumberLen2)
 				pnManager.EXPECT().PopPacketNumber(protocol.Encryption1RTT).Return(protocol.PacketNumber(0x42))
 				sealingManager.EXPECT().Get1RTTSealer().Return(getSealer(), nil)
 				framer.EXPECT().HasData().Return(true)
 				ackFramer.EXPECT().GetAckFrame(protocol.Encryption1RTT, false)
 				frames := []ackhandler.Frame{
-					{Frame: &wire.ResetStreamFrame{}},
+					{Frame: &wire.ResetStreamFrame{}, Handler: &mtuFinderAckHandler{}}, // set any non-nil ackhandler.FrameHandler
 					{Frame: &wire.MaxDataFrame{}},
 				}
 				expectAppendControlFrames(frames...)
@@ -531,9 +528,20 @@ var _ = Describe("Packet packer", func() {
 				p, err := packer.AppendPacket(buffer, maxPacketSize, protocol.Version1)
 				Expect(err).ToNot(HaveOccurred())
 				Expect(p.Frames).To(HaveLen(2))
-				for i, f := range p.Frames {
-					Expect(f).To(BeAssignableToTypeOf(frames[i]))
+				var sawResetStream, sawMaxData bool
+				for _, frame := range p.Frames {
+					switch frame.Frame.(type) {
+					case *wire.ResetStreamFrame:
+						sawResetStream = true
+						Expect(frame.Handler).To(Equal(&mtuFinderAckHandler{}))
+					case *wire.MaxDataFrame:
+						sawMaxData = true
+						Expect(frame.Handler).ToNot(BeNil())
+						Expect(frame.Handler).ToNot(Equal(&mtuFinderAckHandler{}))
+					}
 				}
+				Expect(sawResetStream).To(BeTrue())
+				Expect(sawMaxData).To(BeTrue())
 				Expect(buffer.Len()).ToNot(BeZero())
 			})
 
diff --git a/send_stream_test.go b/send_stream_test.go
index 6279e9ff..0cd032ce 100644
--- a/send_stream_test.go
+++ b/send_stream_test.go
@@ -689,10 +689,7 @@ var _ = Describe("Send Stream", func() {
 	Context("stream cancellations", func() {
 		Context("canceling writing", func() {
 			It("queues a RESET_STREAM frame", func() {
-				gomock.InOrder(
-					mockSender.EXPECT().onHasStreamControlFrame(streamID, gomock.Any()),
-					mockSender.EXPECT().onStreamCompleted(streamID),
-				)
+				mockSender.EXPECT().onHasStreamControlFrame(streamID, gomock.Any())
 				str.writeOffset = 1234
 				str.CancelWrite(9876)
 				cf, ok, hasMore := str.getControlFrame()
@@ -705,6 +702,20 @@ var _ = Describe("Send Stream", func() {
 				Expect(hasMore).To(BeFalse())
 			})
 
+			It("retransmits a RESET_STREAM frame", func() {
+				mockSender.EXPECT().onHasStreamControlFrame(streamID, gomock.Any())
+				str.CancelWrite(9876)
+				cf, ok, _ := str.getControlFrame()
+				Expect(ok).To(BeTrue())
+				Expect(cf.Frame).To(BeAssignableToTypeOf(&wire.ResetStreamFrame{}))
+
+				mockSender.EXPECT().onHasStreamControlFrame(streamID, gomock.Any())
+				cf.Handler.OnLost(cf.Frame)
+				cf2, ok, _ := str.getControlFrame()
+				Expect(ok).To(BeTrue())
+				Expect(cf2.Frame).To(Equal(cf.Frame))
+			})
+
 			// This test is inherently racy, as it tests a concurrent call to Write() and CancelRead().
 			// A single successful run of this test therefore doesn't mean a lot,
 			// for reliable results it has to be run many times.
@@ -753,7 +764,6 @@ var _ = Describe("Send Stream", func() {
 				frame, ok, _ := str.popStreamFrame(50, protocol.Version1)
 				Expect(ok).To(BeTrue())
 				Expect(frame).ToNot(BeNil())
-				mockSender.EXPECT().onStreamCompleted(streamID)
 				str.CancelWrite(1234)
 				Eventually(writeReturned).Should(BeClosed())
 				Expect(n).To(BeEquivalentTo(frame.Frame.DataLen()))
@@ -775,7 +785,6 @@ var _ = Describe("Send Stream", func() {
 				Expect(ok).To(BeTrue())
 				Expect(hasMoreData).To(BeTrue())
 				Expect(frame).ToNot(BeNil())
-				mockSender.EXPECT().onStreamCompleted(streamID)
 				str.CancelWrite(1234)
 				_, ok, hasMoreData = str.popStreamFrame(10, protocol.Version1)
 				Expect(ok).To(BeFalse())
@@ -804,7 +813,6 @@ var _ = Describe("Send Stream", func() {
 				Expect(ok).To(BeTrue())
 				Expect(hasMoreData).To(BeTrue())
 				Expect(frame).ToNot(BeNil())
-				mockSender.EXPECT().onStreamCompleted(streamID)
 				str.CancelWrite(1234)
 				_, ok, hasMoreData = str.popStreamFrame(10, protocol.Version1)
 				Expect(ok).To(BeFalse())
@@ -828,14 +836,12 @@ var _ = Describe("Send Stream", func() {
 				Expect(ok).To(BeTrue())
 				Expect(hasMoreData).To(BeTrue())
 				Expect(frame).ToNot(BeNil())
-				mockSender.EXPECT().onStreamCompleted(streamID)
 				str.CancelWrite(1234)
 				frame.Handler.OnAcked(frame.Frame)
 			})
 
 			It("cancels the context", func() {
 				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				Expect(str.Context().Done()).ToNot(BeClosed())
 				str.CancelWrite(1234)
 				Expect(str.Context().Done()).To(BeClosed())
@@ -845,7 +851,6 @@ var _ = Describe("Send Stream", func() {
 
 			It("doesn't allow further calls to Write", func() {
 				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				str.CancelWrite(1234)
 				_, err := strWithTimeout.Write([]byte("foobar"))
 				Expect(err).To(MatchError(&StreamError{
@@ -857,7 +862,6 @@ var _ = Describe("Send Stream", func() {
 
 			It("only cancels once", func() {
 				mockSender.EXPECT().onHasStreamControlFrame(streamID, gomock.Any())
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				str.CancelWrite(1234)
 				str.CancelWrite(4321)
 				cf, ok, hasMore := str.getControlFrame()
@@ -873,7 +877,6 @@ var _ = Describe("Send Stream", func() {
 			It("queues a RESET_STREAM frame, even if the stream was already closed", func() {
 				mockSender.EXPECT().onHasStreamData(streamID, str)
 				mockSender.EXPECT().onHasStreamControlFrame(streamID, str)
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				Expect(str.Close()).To(Succeed())
 				// don't EXPECT any calls to queueControlFrame
 				str.CancelWrite(123)
@@ -909,7 +912,6 @@ var _ = Describe("Send Stream", func() {
 					ErrorCode: 101,
 				})
 
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				str.CancelWrite(101)
 			})
 
@@ -919,7 +921,6 @@ var _ = Describe("Send Stream", func() {
 				done := make(chan struct{})
 				go func() {
 					defer GinkgoRecover()
-					mockSender.EXPECT().onStreamCompleted(gomock.Any())
 					_, err := str.Write(getData(5000))
 					Expect(err).To(Equal(&StreamError{
 						StreamID:  streamID,
@@ -942,7 +943,6 @@ var _ = Describe("Send Stream", func() {
 					StreamID:  streamID,
 					ErrorCode: 123,
 				})
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				_, err := str.Write([]byte("foobar"))
 				Expect(err).To(Equal(&StreamError{
 					StreamID:  streamID,
@@ -952,12 +952,11 @@ var _ = Describe("Send Stream", func() {
 			})
 
 			It("handles Close after STOP_SENDING", func() {
-				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
+				mockSender.EXPECT().onHasStreamControlFrame(streamID, str)
 				str.handleStopSendingFrame(&wire.StopSendingFrame{
 					StreamID:  streamID,
 					ErrorCode: 123,
 				})
-				mockSender.EXPECT().onStreamCompleted(gomock.Any())
 				str.Close()
 			})
 
@@ -966,10 +965,7 @@ var _ = Describe("Send Stream", func() {
 				str.Close()
 				_, ok, _ := str.popStreamFrame(protocol.MaxByteCount, protocol.Version1)
 				Expect(ok).To(BeTrue())
-				gomock.InOrder(
-					mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any()),
-					mockSender.EXPECT().onStreamCompleted(gomock.Any()),
-				)
+				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
 				str.handleStopSendingFrame(&wire.StopSendingFrame{
 					StreamID:  streamID,
 					ErrorCode: 123,
@@ -979,10 +975,7 @@ var _ = Describe("Send Stream", func() {
 			It("handles STOP_SENDING after Close, but before sending the FIN", func() {
 				mockSender.EXPECT().onHasStreamData(gomock.Any(), gomock.Any())
 				str.Close()
-				gomock.InOrder(
-					mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any()),
-					mockSender.EXPECT().onStreamCompleted(gomock.Any()),
-				)
+				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
 				str.handleStopSendingFrame(&wire.StopSendingFrame{
 					StreamID:  streamID,
 					ErrorCode: 123,
@@ -1093,10 +1086,7 @@ var _ = Describe("Send Stream", func() {
 			Expect(ok).To(BeTrue())
 			Eventually(done).Should(BeClosed())
 			Expect(f).ToNot(BeNil())
-			gomock.InOrder(
-				mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any()),
-				mockSender.EXPECT().onStreamCompleted(streamID),
-			)
+			mockSender.EXPECT().onHasStreamControlFrame(gomock.Any(), gomock.Any())
 			str.CancelWrite(9876)
 			// don't EXPECT any calls to onHasStreamData
 			f.Handler.OnLost(f.Frame)
@@ -1151,6 +1141,36 @@ var _ = Describe("Send Stream", func() {
 			frame.Handler.OnAcked(frame.Frame)
 		})
 
+		It("waits until a RESET_STREAM is acknowledged", func() {
+			mockSender.EXPECT().onHasStreamData(streamID, str)
+			done := make(chan struct{})
+			go func() {
+				defer GinkgoRecover()
+				_, err := strWithTimeout.Write([]byte("foobar"))
+				Expect(err).ToNot(HaveOccurred())
+				close(done)
+			}()
+			waitForWrite()
+
+			frame, ok, hasMoreData := str.popStreamFrame(20, protocol.Version1)
+			Expect(ok).To(BeTrue())
+			Expect(frame.Frame.DataLen()).To(BeEquivalentTo(6))
+			Expect(hasMoreData).To(BeFalse())
+
+			mockSender.EXPECT().onHasStreamControlFrame(streamID, str)
+			str.CancelWrite(1234)
+			cf, ok, _ := str.getControlFrame()
+			Expect(ok).To(BeTrue())
+
+			// Acknowledge the STREAM frame.
+			// This doesn't complete the stream, since we're still waiting for
+			// the acknowledgment of the RESET_STREAM frame.
+			frame.Handler.OnAcked(frame.Frame)
+
+			mockSender.EXPECT().onStreamCompleted(streamID)
+			cf.Handler.OnAcked(cf.Frame)
+		})
+
 		It("says when a stream is completed, if Close() is called before popping the frame", func() {
 			mockSender.EXPECT().onHasStreamData(streamID, str).Times(2)
 			done := make(chan struct{})