Repository: pion/sctp (go)

Issue:
Data channels do not fully close when close initiated from Chrome
In my application where Pion is offering and Chrome is answering:
- Pion application starts a datachannel
- The Javascript in Chrome explicitly closes the datachannel
- Pion receives the datachannel onclose event
- But Chrome's data channel stays in the "closing" state and is never "closed".

Perhaps pion is not doing https://tools.ietf.org/html/draft-ietf-rtcweb-data-channel-13#section-6.7 correctly when the close is initiated remotely?

Using pion/sctp v1.7.11

@Sean-Der suggested I reference @enobufs as you might know if this is expected behaviour

Pull request description:
Fix Chrome datachannel stuck at closing
#### Description

When close datachannel from chrome, sctp should response with an outgoingResetRequest, otherwise the chrome won't fully close the datachannel.

#### Reference issue
Fixes #187 


Patch:
diff --git a/association.go b/association.go
index 0676c48..91b92fc 100644
--- a/association.go
+++ b/association.go
@@ -1985,6 +1985,7 @@ func (a *Association) sendResetRequest(streamIdentifier uint16) error {
 func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 	switch p := raw.(type) {
 	case *paramOutgoingResetRequest:
+		a.log.Tracef("[%s] handleReconfigParam (OutgoingResetRequest)", a.name)
 		a.reconfigRequests[p.reconfigRequestSequenceNumber] = p
 		resp := a.resetStreamsIfAny(p)
 		if resp != nil {
@@ -1993,6 +1994,7 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 		return nil, nil //nolint:nilnil
 
 	case *paramReconfigResponse:
+		a.log.Tracef("[%s] handleReconfigParam (ReconfigResponse)", a.name)
 		delete(a.reconfigs, p.reconfigResponseSequenceNumber)
 		if len(a.reconfigs) == 0 {
 			a.tReconfig.stop()
@@ -2014,7 +2016,11 @@ func (a *Association) resetStreamsIfAny(p *paramOutgoingResetRequest) *packet {
 			if !ok {
 				continue
 			}
-			a.unregisterStream(s, io.EOF)
+			a.lock.Unlock()
+			s.onInboundStreamReset()
+			a.lock.Lock()
+			a.log.Debugf("[%s] deleting stream %d", a.name, id)
+			delete(a.streams, s.streamIdentifier)
 		}
 		delete(a.reconfigRequests, p.reconfigRequestSequenceNumber)
 	} else {
diff --git a/stream.go b/stream.go
index 58041b1..f9304e1 100644
--- a/stream.go
+++ b/stream.go
@@ -20,9 +20,32 @@ const (
 	ReliabilityTypeTimed byte = 2
 )
 
+// StreamState is an enum for SCTP Stream state field
+// This field identifies the state of stream.
+type StreamState int
+
+// StreamState enums
+const (
+	StreamStateOpen    StreamState = iota // Stream object starts with StreamStateOpen
+	StreamStateClosing                    // Outgoing stream is being reset
+	StreamStateClosed                     // Stream has been closed
+)
+
+func (ss StreamState) String() string {
+	switch ss {
+	case StreamStateOpen:
+		return "open"
+	case StreamStateClosing:
+		return "closing"
+	case StreamStateClosed:
+		return "closed"
+	}
+	return "unknown"
+}
+
 var (
 	errOutboundPacketTooLarge = errors.New("outbound packet larger than maximum message size")
-	errStreamClosed           = errors.New("Stream closed")
+	errStreamClosed           = errors.New("stream closed")
 )
 
 // Stream represents an SCTP stream
@@ -35,13 +58,13 @@ type Stream struct {
 	sequenceNumber      uint16
 	readNotifier        *sync.Cond
 	readErr             error
-	writeErr            error
 	unordered           bool
 	reliabilityType     byte
 	reliabilityValue    uint32
 	bufferedAmount      uint64
 	bufferedAmountLow   uint64
 	onBufferedAmountLow func()
+	state               StreamState
 	log                 logging.LeveledLogger
 	name                string
 }
@@ -178,32 +201,23 @@ func (s *Stream) Write(p []byte) (n int, err error) {
 }
 
 // WriteSCTP writes len(p) bytes from p to the DTLS connection
-func (s *Stream) WriteSCTP(p []byte, ppi PayloadProtocolIdentifier) (n int, err error) {
+func (s *Stream) WriteSCTP(p []byte, ppi PayloadProtocolIdentifier) (int, error) {
 	maxMessageSize := s.association.MaxMessageSize()
 	if len(p) > int(maxMessageSize) {
 		return 0, fmt.Errorf("%w: %v", errOutboundPacketTooLarge, math.MaxUint16)
 	}
 
-	switch s.association.getState() {
-	case shutdownSent, shutdownAckSent, shutdownPending, shutdownReceived:
-		s.lock.Lock()
-		if s.writeErr == nil {
-			s.writeErr = errStreamClosed
-		}
-		s.lock.Unlock()
-	default:
+	if s.State() != StreamStateOpen {
+		return 0, errStreamClosed
 	}
 
-	s.lock.RLock()
-	err = s.writeErr
-	s.lock.RUnlock()
+	chunks := s.packetize(p, ppi)
+	n := len(p)
+	err := s.association.sendPayloadData(chunks)
 	if err != nil {
-		return 0, err
+		return n, errStreamClosed
 	}
-
-	chunks := s.packetize(p, ppi)
-
-	return len(p), s.association.sendPayloadData(chunks)
+	return n, nil
 }
 
 func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPayloadData {
@@ -267,26 +281,23 @@ func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPa
 // Close closes the write-direction of the stream.
 // Future calls to Write are not permitted after calling Close.
 func (s *Stream) Close() error {
-	if sid, isOpen := func() (uint16, bool) {
+	if sid, resetOutbound := func() (uint16, bool) {
 		s.lock.Lock()
 		defer s.lock.Unlock()
 
-		isOpen := true
-		if s.writeErr == nil {
-			s.writeErr = errStreamClosed
-		} else {
-			isOpen = false
-		}
+		s.log.Debugf("[%s] Close: state=%s", s.name, s.state.String())
 
-		if s.readErr == nil {
-			s.readErr = io.EOF
-		} else {
-			isOpen = false
+		if s.state == StreamStateOpen {
+			if s.readErr == nil {
+				s.state = StreamStateClosing
+			} else {
+				s.state = StreamStateClosed
+			}
+			s.log.Debugf("[%s] state change: open => %s", s.name, s.state.String())
+			return s.streamIdentifier, true
 		}
-		s.readNotifier.Broadcast() // broadcast regardless
-
-		return s.streamIdentifier, isOpen
-	}(); isOpen {
+		return s.streamIdentifier, false
+	}(); resetOutbound {
 		// Reset the outgoing stream
 		// https://tools.ietf.org/html/rfc6525
 		return s.association.sendResetRequest(sid)
@@ -365,3 +376,35 @@ func (s *Stream) getNumBytesInReassemblyQueue() int {
 	// No lock is required as it reads the size with atomic load function.
 	return s.reassemblyQueue.getNumBytes()
 }
+
+func (s *Stream) onInboundStreamReset() {
+	s.lock.Lock()
+	defer s.lock.Unlock()
+
+	s.log.Debugf("[%s] onInboundStreamReset: state=%s", s.name, s.state.String())
+
+	// No more inbound data to read. Unblock the read with io.EOF.
+	// This should cause DCEP layer (datachannel package) to call Close() which
+	// will reset outgoing stream also.
+
+	// See RFC 8831 section 6.7:
+	//	if one side decides to close the data channel, it resets the corresponding
+	//	outgoing stream.  When the peer sees that an incoming stream was
+	//	reset, it also resets its corresponding outgoing stream.  Once this
+	//	is completed, the data channel is closed.
+
+	s.readErr = io.EOF
+	s.readNotifier.Broadcast()
+
+	if s.state == StreamStateClosing {
+		s.log.Debugf("[%s] state change: closing => closed", s.name)
+		s.state = StreamStateClosed
+	}
+}
+
+// State return the stream state.
+func (s *Stream) State() StreamState {
+	s.lock.RLock()
+	defer s.lock.RUnlock()
+	return s.state
+}


Test patch:
diff --git a/vnet_test.go b/vnet_test.go
index bdf7994..da7af14 100644
--- a/vnet_test.go
+++ b/vnet_test.go
@@ -1,6 +1,7 @@
 package sctp
 
 import (
+	"bytes"
 	"fmt"
 	"math/rand"
 	"net"
@@ -386,84 +387,91 @@ func TestRwndFull(t *testing.T) {
 }
 
 func TestStreamClose(t *testing.T) {
-	lim := test.TimeOut(time.Second * 10)
-	defer lim.Stop()
-
-	loggerFactory := logging.NewDefaultLoggerFactory()
-	log := loggerFactory.NewLogger("test")
-
-	venv, err := buildVNetEnv(&vNetEnvConfig{
-		loggerFactory: loggerFactory,
-		log:           log,
-	})
-	if !assert.NoError(t, err, "should succeed") {
-		return
-	}
-	if !assert.NotNil(t, venv, "should not be nil") {
-		return
-	}
-	defer venv.wan.Stop() // nolint:errcheck
+	loopBackTest := func(t *testing.T, dropReconfigChunk bool) {
+		lim := test.TimeOut(time.Second * 10)
+		defer lim.Stop()
 
-	serverStreamReady := make(chan struct{})
-	clientStreamReady := make(chan struct{})
-	clientStartClose := make(chan struct{})
-	serverStreamClosed := make(chan struct{})
-	shutDownClient := make(chan struct{})
-	clientShutDown := make(chan struct{})
-	serverShutDown := make(chan struct{})
+		loggerFactory := logging.NewDefaultLoggerFactory()
+		log := loggerFactory.NewLogger("test")
 
-	go func() {
-		defer close(serverShutDown)
-		// connected UDP conn for server
-		conn, err := venv.net0.DialUDP("udp4",
-			&net.UDPAddr{IP: net.ParseIP("1.1.1.1"), Port: 5000},
-			&net.UDPAddr{IP: net.ParseIP("2.2.2.2"), Port: 5000},
-		)
+		venv, err := buildVNetEnv(&vNetEnvConfig{
+			loggerFactory: loggerFactory,
+			log:           log,
+		})
 		if !assert.NoError(t, err, "should succeed") {
 			return
 		}
-		defer conn.Close() // nolint:errcheck
-
-		// server association
-		assoc, err := Server(Config{
-			NetConn:       conn,
-			LoggerFactory: loggerFactory,
-		})
-		if !assert.NoError(t, err, "should succeed") {
+		if !assert.NotNil(t, venv, "should not be nil") {
 			return
 		}
-		defer assoc.Close() // nolint:errcheck
+		defer venv.wan.Stop() // nolint:errcheck
 
-		log.Info("server handshake complete")
+		clientShutDown := make(chan struct{})
+		serverShutDown := make(chan struct{})
 
-		stream, err := assoc.AcceptStream()
-		if !assert.NoError(t, err, "should succeed") {
-			return
+		const numMessages = 10
+		const messageSize = 1024
+		var messages [][]byte
+		var numServerReceived int
+		var numClientReceived int
+
+		for i := 0; i < numMessages; i++ {
+			bytes := make([]byte, messageSize)
+			messages = append(messages, bytes)
 		}
-		defer stream.Close() // nolint:errcheck
 
-		buf := make([]byte, 1500)
-		for {
-			n, err := stream.Read(buf)
-			if err != nil {
-				t.Logf("server: Read returned %v", err)
-				break
+		go func() {
+			defer close(serverShutDown)
+			// connected UDP conn for server
+			conn, innerErr := venv.net0.DialUDP("udp4",
+				&net.UDPAddr{IP: net.ParseIP("1.1.1.1"), Port: 5000},
+				&net.UDPAddr{IP: net.ParseIP("2.2.2.2"), Port: 5000},
+			)
+			if !assert.NoError(t, innerErr, "should succeed") {
+				return
+			}
+			defer conn.Close() // nolint:errcheck
+
+			// server association
+			assoc, innerErr := Server(Config{
+				NetConn:       conn,
+				LoggerFactory: loggerFactory,
+			})
+			if !assert.NoError(t, innerErr, "should succeed") {
+				return
 			}
+			defer assoc.Close() // nolint:errcheck
+
+			log.Info("server handshake complete")
 
-			if !assert.Equal(t, "HELLO", string(buf[:n]), "should receive HELLO") {
-				continue
+			stream, innerErr := assoc.AcceptStream()
+			if !assert.NoError(t, innerErr, "should succeed") {
+				return
 			}
+			assert.Equal(t, StreamStateOpen, stream.State())
 
-			log.Info("server stream ready")
-			close(serverStreamReady)
-		}
+			buf := make([]byte, 1500)
+			for {
+				n, errRead := stream.Read(buf)
+				if errRead != nil {
+					log.Infof("server: Read returned %v", errRead)
+					_ = stream.Close() // nolint:errcheck
+					assert.Equal(t, StreamStateClosed, stream.State())
+					break
+				}
 
-		close(serverStreamClosed)
-		log.Info("server closing")
-	}()
+				log.Infof("server: received %d bytes (%d)", n, numServerReceived)
+				assert.Equal(t, 0, bytes.Compare(buf[:n], messages[numServerReceived]), "should receive HELLO")
+
+				_, err = stream.Write(buf[:n])
+				assert.NoError(t, err, "should succeed")
+
+				numServerReceived++
+			}
+			// don't close association until the client's stream routine is complete
+			<-clientShutDown
+		}()
 
-	go func() {
-		defer close(clientShutDown)
 		// connected UDP conn for client
 		conn, err := venv.net1.DialUDP("udp4",
 			&net.UDPAddr{IP: net.ParseIP("2.2.2.2"), Port: 5000},
@@ -472,6 +480,7 @@ func TestStreamClose(t *testing.T) {
 		if !assert.NoError(t, err, "should succeed") {
 			return
 		}
+		defer conn.Close() // nolint:errcheck
 
 		// client association
 		assoc, err := Client(Config{
@@ -489,45 +498,51 @@ func TestStreamClose(t *testing.T) {
 		if !assert.NoError(t, err, "should succeed") {
 			return
 		}
-
+		assert.Equal(t, StreamStateOpen, stream.State())
 		stream.SetReliabilityParams(false, ReliabilityTypeReliable, 0)
 
-		// Send a message to let server side stream to open
-		_, err = stream.Write([]byte("HELLO"))
-		if !assert.NoError(t, err, "should succeed") {
-			return
-		}
-
+		// begin client read-loop
 		buf := make([]byte, 1500)
-		done := make(chan struct{})
 		go func() {
+			defer close(clientShutDown)
 			for {
-				log.Info("client read")
-				_, err2 := stream.Read(buf)
+				n, err2 := stream.Read(buf)
 				if err2 != nil {
-					t.Logf("client: Read returned %v", err2)
+					log.Infof("client: Read returned %v", err2)
+					assert.Equal(t, StreamStateClosed, stream.State())
 					break
 				}
+
+				log.Infof("client: received %d bytes (%d)", n, numClientReceived)
+				assert.Equal(t, 0, bytes.Compare(buf[:n], messages[numClientReceived]), "should receive HELLO")
+				numClientReceived++
 			}
-			close(done)
 		}()
 
-		log.Info("client stream ready")
-		close(clientStreamReady)
-
-		<-clientStartClose
+		// Send messages to the server
+		for i := 0; i < numMessages; i++ {
+			_, err = stream.Write(messages[i])
+			assert.NoError(t, err, "should succeed")
+		}
 
-		// drop next 1 RECONFIG chunk
-		venv.dropNextReconfigChunk(1)
+		if dropReconfigChunk {
+			venv.dropNextReconfigChunk(1)
+		}
 
+		// Immediately close the stream
 		err = stream.Close()
 		assert.NoError(t, err, "should succeed")
+		assert.Equal(t, StreamStateClosing, stream.State())
 
 		log.Info("client wait for exit reading..")
-		<-done
+		<-clientShutDown
 
-		<-shutDownClient
+		assert.Equal(t, numMessages, numServerReceived, "all messages should be received")
+		assert.Equal(t, numMessages, numClientReceived, "all messages should be received")
+
+		_, err = stream.Write([]byte{1})
 
+		assert.Equal(t, err, errStreamClosed, "after closed should not allow write")
 		// Check if RECONFIG was actually dropped
 		assert.Equal(t, 0, venv.numToDropReconfig, "should be zero")
 
@@ -539,26 +554,15 @@ func TestStreamClose(t *testing.T) {
 		pendingReconfigs := len(assoc.reconfigs)
 		assoc.lock.RUnlock()
 		assert.Equal(t, 0, pendingReconfigs, "should be zero")
+	}
 
-		log.Info("client closing")
-	}()
-
-	// wait until both establish a stream
-	<-clientStreamReady
-	<-serverStreamReady
-
-	log.Info("stream ready")
-
-	// let client begin writing
-	log.Info("client start closing")
-	close(clientStartClose)
-
-	<-serverStreamClosed
-	close(shutDownClient)
+	t.Run("without dropping Reconfig", func(t *testing.T) {
+		loopBackTest(t, false)
+	})
 
-	<-clientShutDown
-	<-serverShutDown
-	log.Info("all done")
+	t.Run("with dropping Reconfig", func(t *testing.T) {
+		loopBackTest(t, true)
+	})
 }
 
 // this test case reproduces the issue mentioned in