Repository: pion/sctp
Pull request #255 (2022-12-12 21:13:22)

Revert "Remove pending buffer when stream closed"
This reverts commit c0159aa2d49c240362038edf88baa8a9e6cfcede.

@jerry-tao @enobufs, the reverted commit made sense in theory but it's causing a regression in what https://github.com/pion/webrtc appears to expect in terms of data channel closing behavior with respect to pending writes on the sender side. More specifically, `TestEOF/Detach` started to fail where we expected that when the sender did a write followed by close, that the reader would see those writes and then see an EOF. That is to say, the enqueued writes are lost due to these changes. It doesn't seem like this change fixed any correctness issues with regards to the spec so it seems like we should revert it so as to unblock other changes in this v1.8.x. v1.8.4 is currently published but is now broken/flaky. https://github.com/pion/sctp/pull/254 attempted to fix this but it introduced more issues due to how we lose a lot of context when we `Close` in that we don't know why we are closing (inbound reset or user requested).

codecov[bot]: # [Codecov](https://codecov.io/gh/pion/sctp/pull/255?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) Report
Base: **80.74**% // Head: **80.92**% // Increases project coverage by **`+0.18%`** :tada:
> Coverage data is based on head [(`6b3e43d`)](https://codecov.io/gh/pion/sctp/pull/255?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) compared to base [(`6e962c6`)](https://codecov.io/gh/pion/sctp/commit/6e962c66d1fe66ec4347f35b000d28f010443778?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).
> Patch coverage: 85.71% of modified lines in pull request are covered.

<details><summary>Additional details and impacted files</summary>


```diff
@@            Coverage Diff             @@
##           master     #255      +/-   ##
==========================================
+ Coverage   80.74%   80.92%   +0.18%     
==========================================
  Files          48       48              
  Lines        4046     4032      -14     
==========================================
- Hits         3267     3263       -4     
+ Misses        636      628       -8     
+ Partials      143      141       -2     
```

| Flag | Coverage Δ | |
|---|---|---|
| go | `80.92% <85.71%> (+0.18%)` | :arrow_up: |
| wasm | `67.46% <85.71%> (+0.92%)` | :arrow_up: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#carryforward-flags-in-the-pull-request-comment) to find out more.

| [Impacted Files](https://codecov.io/gh/pion/sctp/pull/255?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion) | Coverage Δ | |
|---|---|---|
| [chunk\_payload\_data.go](https://codecov.io/gh/pion/sctp/pull/255/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-Y2h1bmtfcGF5bG9hZF9kYXRhLmdv) | `72.83% <ø> (ø)` | |
| [association.go](https://codecov.io/gh/pion/sctp/pull/255/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-YXNzb2NpYXRpb24uZ28=) | `84.91% <25.00%> (+0.21%)` | :arrow_up: |
| [stream.go](https://codecov.io/gh/pion/sctp/pull/255/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion#diff-c3RyZWFtLmdv) | `92.70% <100.00%> (+1.55%)` | :arrow_up: |

Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion). Have a feature suggestion? [Share it here.](https://app.codecov.io/gh/feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion)

</details>

[:umbrella: View full report at Codecov](https://codecov.io/gh/pion/sctp/pull/255?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).   
:loudspeaker: Do you have feedback about the report comment? [Let us know in this issue](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pion).

edaniels: I reverted for one commit and back in `master` to unblock other feature work.
edaniels: If we're okay with how this is breaking the `webrtc` tests, then I think we should just go update the `TestEOF/Detach` test in `datachannel_go_test.go` to use channel synchronization to wait for the read of `this is some test data` to happen prior to the sender side closing; otherwise the test will be flaky.
jerry-tao: LGTM.
jerry-tao: As we discussed in #245, the commit does not fully resolve the origin issue either. As I said in #239 , the pion/sctp is more like webrtc/sctp_transport+webrtc/datachannel+libsrtcp.
Maybe in v2.0 we could split the sctp_transport and datachannel logic to pion/datachannel.
As for the correctness, this bothers me too, should we have the same logic with C++ webrtc/src or implements it according to RFC with personal understanding?

Patch:
diff --git a/association.go b/association.go
index bf8f21d..91b92fc 100644
--- a/association.go
+++ b/association.go
@@ -221,9 +221,8 @@ type Association struct {
 	delayedAckTriggered   bool
 	immediateAckTriggered bool
 
-	name          string
-	log           logging.LeveledLogger
-	streamVersion uint32
+	name string
+	log  logging.LeveledLogger
 }
 
 // Config collects the arguments to createAssociation construction into
@@ -1369,7 +1368,6 @@ func (a *Association) createStream(streamIdentifier uint16, accept bool) *Stream
 		streamIdentifier: streamIdentifier,
 		reassemblyQueue:  newReassemblyQueue(streamIdentifier),
 		log:              a.log,
-		version:          atomic.AddUint32(&a.streamVersion, 1),
 		name:             fmt.Sprintf("%d:%s", streamIdentifier, a.name),
 	}
 
@@ -2090,14 +2088,10 @@ func (a *Association) popPendingDataChunksToSend() ([]*chunkPayloadData, []uint1
 			dataLen := uint32(len(c.userData))
 			if dataLen == 0 {
 				sisToReset = append(sisToReset, c.streamIdentifier)
-				a.popPendingDataChunksToDrop(c)
-				continue
-			}
-
-			s, ok := a.streams[c.streamIdentifier]
-
-			if !ok || s.State() > StreamStateOpen || s.version != c.streamVersion {
-				a.popPendingDataChunksToDrop(c)
+				err := a.pendingQueue.pop(c)
+				if err != nil {
+					a.log.Errorf("failed to pop from pending queue: %s", err.Error())
+				}
 				continue
 			}
 
@@ -2129,13 +2123,6 @@ func (a *Association) popPendingDataChunksToSend() ([]*chunkPayloadData, []uint1
 	return chunks, sisToReset
 }
 
-func (a *Association) popPendingDataChunksToDrop(c *chunkPayloadData) {
-	err := a.pendingQueue.pop(c)
-	if err != nil {
-		a.log.Errorf("failed to pop from pending queue: %s", err.Error())
-	}
-}
-
 // bundleDataChunksIntoPackets packs DATA chunks into packets. It tries to bundle
 // DATA chunks into a packet so long as the resulting packet size does not exceed
 // the path MTU.
diff --git a/chunk_payload_data.go b/chunk_payload_data.go
index 294de8a..eecc7c8 100644
--- a/chunk_payload_data.go
+++ b/chunk_payload_data.go
@@ -71,8 +71,7 @@ type chunkPayloadData struct {
 	// chunk is still in the inflight queue
 	retransmit bool
 
-	head          *chunkPayloadData // link to the head of the fragment
-	streamVersion uint32
+	head *chunkPayloadData // link to the head of the fragment
 }
 
 const (
diff --git a/stream.go b/stream.go
index c1fbece..fd2c5fc 100644
--- a/stream.go
+++ b/stream.go
@@ -24,12 +24,12 @@ const (
 
 // StreamState is an enum for SCTP Stream state field
 // This field identifies the state of stream.
-type StreamState int32
+type StreamState int
 
 // StreamState enums
 const (
 	StreamStateOpen    StreamState = iota // Stream object starts with StreamStateOpen
-	StreamStateClosing                    // Stream is closed by remote
+	StreamStateClosing                    // Outgoing stream is being reset
 	StreamStateClosed                     // Stream has been closed
 )
 
@@ -71,7 +71,6 @@ type Stream struct {
 	state               StreamState
 	log                 logging.LeveledLogger
 	name                string
-	version             uint32
 }
 
 // StreamIdentifier returns the Stream identifier associated to the stream.
@@ -297,7 +296,6 @@ func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPa
 		copy(userData, raw[i:i+fragmentSize])
 
 		chunk := &chunkPayloadData{
-			streamVersion:        s.version,
 			streamIdentifier:     s.streamIdentifier,
 			userData:             userData,
 			unordered:            unordered,
@@ -340,22 +338,16 @@ func (s *Stream) Close() error {
 		s.lock.Lock()
 		defer s.lock.Unlock()
 
-		state := s.State()
-		s.log.Debugf("[%s] Close: state=%s", s.name, state.String())
+		s.log.Debugf("[%s] Close: state=%s", s.name, s.state.String())
 
-		switch state {
-		case StreamStateOpen:
-			s.SetState(StreamStateClosed)
-			s.log.Debugf("[%s] state change: open => closed", s.name)
-			s.readErr = io.EOF
-			s.readNotifier.Broadcast()
-			return s.streamIdentifier, true
-		case StreamStateClosing:
-			s.SetState(StreamStateClosed)
-			s.log.Debugf("[%s] state change: closing => closed", s.name)
+		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
-		case StreamStateClosed:
-			return s.streamIdentifier, false
 		}
 		return s.streamIdentifier, false
 	}(); resetOutbound {
@@ -442,8 +434,7 @@ func (s *Stream) onInboundStreamReset() {
 	s.lock.Lock()
 	defer s.lock.Unlock()
 
-	state := s.State()
-	s.log.Debugf("[%s] onInboundStreamReset: state=%s", s.name, state.String())
+	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
@@ -454,21 +445,19 @@ func (s *Stream) onInboundStreamReset() {
 	//	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.
-	if state == StreamStateOpen {
-		s.log.Debugf("[%s] state change: open => closing", s.name)
-		s.SetState(StreamStateClosing)
-	}
 
 	s.readErr = io.EOF
 	s.readNotifier.Broadcast()
-}
 
-// State atomically returns the stream state.
-func (s *Stream) State() StreamState {
-	return StreamState(atomic.LoadInt32((*int32)(&s.state)))
+	if s.state == StreamStateClosing {
+		s.log.Debugf("[%s] state change: closing => closed", s.name)
+		s.state = StreamStateClosed
+	}
 }
 
-// SetState atomically sets the stream state.
-func (s *Stream) SetState(newState StreamState) {
-	atomic.StoreInt32((*int32)(&s.state), int32(newState))
+// 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..996f495 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")
+
+				_, err2 := stream.Write(buf[:n])
+				assert.NoError(t, err2, "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