don't cancel streams after shutdown (#4673)

This ensures that `stream.Write` and `stream.Read` return the error code
from connection close, if the stream was closed as a result of
connection close.

receive_stream.go

@@ -253,6 +253,9 @@ func (s *receiveStream) cancelReadImpl(errorCode qerr.StreamErrorCode) (queuedNe
 	if s.cancelledLocally { // duplicate call to CancelRead
 		return false
 	}
+	if s.closeForShutdownErr != nil {
+		return false
+	}
 	s.cancelledLocally = true
 	if s.errorRead || s.cancelledRemotely {
 		return false

receive_stream_test.go

@@ -630,6 +630,17 @@ var _ = Describe("Receive Stream", func() {
 					Fin:    true,
 				})).To(Succeed())
 			})
+
+			It("ignores cancellations after closeForShutdown", func() {
+				closeErr := errors.New("closed for shutdown")
+				str.closeForShutdown(closeErr)
+				buf := make([]byte, 100)
+				_, err := str.Read(buf)
+				Expect(err).To(Equal(closeErr))
+				str.CancelRead(42)
+				_, err = str.Read(buf)
+				Expect(err).To(Equal(closeErr))
+			})
 		})
 
 		Context("receiving RESET_STREAM frames", func() {

send_stream.go

@@ -423,6 +423,10 @@ func (s *sendStream) CancelWrite(errorCode StreamErrorCode) {
 
 func (s *sendStream) cancelWriteImpl(errorCode qerr.StreamErrorCode, remote bool) {
 	s.mutex.Lock()
+	if s.closeForShutdownErr != nil {
+		s.mutex.Unlock()
+		return
+	}
 	if !remote {
 		s.cancellationFlagged = true
 		if s.cancelWriteErr != nil {

send_stream_test.go

@@ -981,6 +981,15 @@ var _ = Describe("Send Stream", func() {
 					ErrorCode: 123,
 				})
 			})
+			It("ignores cancellations after closeForShutdown", func() {
+				closeErr := errors.New("closed for shutdown")
+				str.closeForShutdown(closeErr)
+				_, err := str.Write([]byte("hello"))
+				Expect(err).To(Equal(closeErr))
+				str.CancelWrite(42)
+				_, err = str.Write([]byte("hello"))
+				Expect(err).To(Equal(closeErr))
+			})
 		})
 	})