don't enqueue stream for sending on reordered MAX_STREAM_DATA frames (#…

…4269)

internal/flowcontrol/base_flow_controller.go

@@ -48,10 +48,12 @@ func (c *baseFlowController) AddBytesSent(n protocol.ByteCount) {
 }
 
 // UpdateSendWindow is called after receiving a MAX_{STREAM_}DATA frame.
-func (c *baseFlowController) UpdateSendWindow(offset protocol.ByteCount) {
+func (c *baseFlowController) UpdateSendWindow(offset protocol.ByteCount) (updated bool) {
  if offset > c.sendWindow {
  c.sendWindow = offset
+ return true
  }
+ return false
 }
 
 func (c *baseFlowController) sendWindowSize() protocol.ByteCount {

internal/flowcontrol/base_flow_controller_test.go

@@ -59,9 +59,9 @@ var _ = Describe("Base Flow controller", func() {
  })
 
  It("does not decrease the flow control window", func() {
- controller.UpdateSendWindow(20)
+ Expect(controller.UpdateSendWindow(20)).To(BeTrue())
  Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(20)))
- controller.UpdateSendWindow(10)
+ Expect(controller.UpdateSendWindow(10)).To(BeFalse())
  Expect(controller.sendWindowSize()).To(Equal(protocol.ByteCount(20)))
  })
 

internal/flowcontrol/interface.go

@@ -5,7 +5,7 @@ import "github.com/quic-go/quic-go/internal/protocol"
 type flowController interface {
  // for sending
  SendWindowSize() protocol.ByteCount
- UpdateSendWindow(protocol.ByteCount)
+ UpdateSendWindow(protocol.ByteCount) (updated bool)
  AddBytesSent(protocol.ByteCount)
  // for receiving
  AddBytesRead(protocol.ByteCount)
@@ -16,12 +16,11 @@ type flowController interface {
 // A StreamFlowController is a flow controller for a QUIC stream.
 type StreamFlowController interface {
  flowController
- // for receiving
- // UpdateHighestReceived should be called when a new highest offset is received
+ // 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
  UpdateHighestReceived(offset protocol.ByteCount, final bool) error
- // Abandon should be called when reading from the stream is aborted early,
+ // Abandon is called when reading from the stream is aborted early,
  // and there won't be any further calls to AddBytesRead.
  Abandon()
 }

send_stream.go

@@ -404,11 +404,13 @@ func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool
 }
 
 func (s *sendStream) updateSendWindow(limit protocol.ByteCount) {
+ updated := s.flowController.UpdateSendWindow(limit)
+ if !updated { // duplicate or reordered MAX_STREAM_DATA frame
+ return
+ }
  s.mutex.Lock()
  hasStreamData := s.dataForWriting != nil || s.nextFrame != nil
  s.mutex.Unlock()
-
- s.flowController.UpdateSendWindow(limit)
  if hasStreamData {
  s.sender.onHasStreamData(s.streamID)
  }

send_stream_test.go

@@ -682,7 +682,7 @@ var _ = Describe("Send Stream", func() {
  })
 
  It("says when it has data for sending", func() {
- mockFC.EXPECT().UpdateSendWindow(gomock.Any())
+ mockFC.EXPECT().UpdateSendWindow(gomock.Any()).Return(true)
  mockSender.EXPECT().onHasStreamData(streamID)
  done := make(chan struct{})
  go func() {
@@ -698,6 +698,24 @@ var _ = Describe("Send Stream", func() {
  str.closeForShutdown(nil)
  Eventually(done).Should(BeClosed())
  })
+
+ It("doesn't say it has data for sending if the MAX_STREAM_DATA frame was reordered", func() {
+ mockFC.EXPECT().UpdateSendWindow(gomock.Any()).Return(false) // reordered frame
+ mockSender.EXPECT().onHasStreamData(streamID)
+ done := make(chan struct{})
+ go func() {
+ defer GinkgoRecover()
+ _, err := str.Write([]byte("foobar"))
+ Expect(err).ToNot(HaveOccurred())
+ close(done)
+ }()
+ waitForWrite()
+ // don't expect any calls to onHasStreamData
+ str.updateSendWindow(42)
+ // make sure the Write go routine returns
+ str.closeForShutdown(nil)
+ Eventually(done).Should(BeClosed())
+ })
  })
 
  Context("stream cancellations", func() {