Repository: pion/sctp
Pull request #107 (2020-02-17 06:56:05)

Fix ProtocolViolation error from usrsctp and rwnd exhaustion
This PR fixes two issues:
* #104 Abort (ProtocolViolation) with FwdTSN for fragmented messages
* #106  Receive buffer leakage on FwdTSN for unordered data chunks

Test:
* pion/sctp unit tests pass with no race
* Ordered/Unordered data channel test (manual) with Chrome and Firefox

This PR replaces earlier PR #105.

mturnshek: @enobufs 

I can confirm this works way better than master. I can send 50000 bytes messages at 50fps with 10% packet loss and 150ms latency introduced on UDP loopback for a solid minute before data channels close. On master, the data channels close instantly with that load and loss introduction. So I think you really solved 99% of the problems people will run into already.

However, on this branch with the latest commit, they still close with that same ABORT line:

```
sctp TRACE: 12:07:47.147089 association.go:1525: [0xc000356000] building fwdtsn: newCumulativeTSN=1205033939 - (si=4 ssn=619)
sctp DEBUG: 12:07:47.167431 association.go:1982: [0xc000356000] Abort chunk, with following errors: (Protocol Violation: Reass 30000007,CI:ffffffff,TSN=47d35bc6,SID=0004,FSN=47d35bc6,SSN:026b)
sctp DEBUG: 12:07:47.168489 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.168540 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.168928 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.169004 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.169059 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.169137 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.169182 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.169536 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.170238 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.170527 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.170669 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171003 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171081 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171643 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171697 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171739 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.171793 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
sctp DEBUG: 12:07:47.172002 association.go:1982: [0xc000356000] Abort chunk, with following errors: 
```

This is the full Pion trace of the latest closing where I received that message: https://gist.github.com/mturnshek/3308f0d0646415af82269c15295af4e9

---

Can I help you reproduce more consistently? I can reproduce ABORTs very reliably with the following setup:

JS datachannels: https://gist.github.com/mturnshek/c19c3c9c8d34ed2ac93d98076fcc953b
Go datachannels: https://gist.github.com/mturnshek/7c6e2bd46e6c1517a91ba649f9c8a730
(these gists are a bit hacky, but are meant to mirror and amplify my production use)

These gists are based on the following example:
https://github.com/pion/webrtc/tree/master/examples/data-channels
https://jsfiddle.net/4ra8w1qe/ (I use this JSFiddle to run the JS gist above)

To reproduce:
- I open the JSFiddle in chrome, and paste the JS gist in and run.
- I place the Go file in the `examples/data-channels` folder and name it `modified.go`
- Set sdp environment variable and `go build && echo $sdp | PION_LOG_TRACE=all ./modified`
- Connect by copy + pasting signals

I use this tool to introduce packet loss: https://github.com/tylertreat/comcast
with the following commands:
`comcast --device=lo --latency=150 --packet-loss=10% --target-proto=udp`
`comcast --device lo --stop

I usually adjust the largeSize and Ticker time in the `modified.go` file, and vary the packet loss and latency to profile different branches.

Using this test, I have collected some ABORTs even on the `disable-forward-tsn` branch. As far as I can tell, these are not `Protocol Violation: Reass 30000007` because I just receive an empty string on the `disable-forward-tsn` branch.
mturnshek: Given that ABORTs are far less common on this branch than on master, it still likely makes sense to get this in before tracking down the more edgy cases.
Sean-Der: @mturnshek thank you so much for the testing! I will reproduce and get the exact chunks that caused the error, should help track down the issue :)
Sean-Der: The `disable-forward-tsn` probably kicked up some other weird behaviors :/ I think that branch is at best a temporary stopgap. Probably not worth using/debugging any further.
enobufs: @Sean-Der @mturnshek Thanks so much for your help.
This PR includes issue #106. Per @mturnshek 's test indicating the improvement (not perfect, however), let me land this on the master, then I will look into the issue further afterward.

Patch:
diff --git a/association.go b/association.go
index e47195d..0fb786d 100644
--- a/association.go
+++ b/association.go
@@ -597,7 +597,7 @@ func (a *Association) gatherOutboundFrastRetransmissionPackets(rawPackets [][]by
 				break // end of pending data
 			}
 
-			if c.acked || c.abandoned {
+			if c.acked || c.abandoned() {
 				continue
 			}
 
@@ -1167,8 +1167,8 @@ func (a *Association) getOrCreateStream(streamIdentifier uint16) *Stream {
 }
 
 // The caller should hold the lock.
-func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, error) {
-	var totalBytesAcked int
+func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (map[uint16]int, uint32, error) {
+	bytesAckedPerStream := map[uint16]int{}
 
 	// New ack point, so pop all ACKed packets from inflightQueue
 	// We add 1 because the "currentAckPoint" has already been popped from the inflight queue
@@ -1176,7 +1176,7 @@ func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, er
 	for i := a.cumulativeTSNAckPoint + 1; sna32LTE(i, d.cumulativeTSNAck); i++ {
 		c, ok := a.inflightQueue.pop(i)
 		if !ok {
-			return 0, 0, errors.Errorf("tsn %v unable to be popped from inflight queue", i)
+			return nil, 0, errors.Errorf("tsn %v unable to be popped from inflight queue", i)
 		}
 
 		if !c.acked {
@@ -1192,16 +1192,13 @@ func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, er
 
 			nBytesAcked := len(c.userData)
 
-			// Report the number of bytes acknowledged to the stream who sent this DATA
-			// chunk.
-			if s, ok := a.streams[c.streamIdentifier]; ok {
-				a.lock.Unlock()
-				s.onBufferReleased(nBytesAcked)
-				a.lock.Lock()
+			// Sum the number of bytes acknowledged per stream
+			if amount, ok := bytesAckedPerStream[c.streamIdentifier]; ok {
+				bytesAckedPerStream[c.streamIdentifier] = amount + nBytesAcked
+			} else {
+				bytesAckedPerStream[c.streamIdentifier] = nBytesAcked
 			}
 
-			totalBytesAcked += nBytesAcked
-
 			// RFC 4960 sec 6.3.1.  RTO Calculation
 			//   C4)  When data is in flight and when allowed by rule C5 below, a new
 			//        RTT measurement MUST be made each round trip.  Furthermore, new
@@ -1234,22 +1231,19 @@ func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, er
 			tsn := d.cumulativeTSNAck + uint32(i)
 			c, ok := a.inflightQueue.get(tsn)
 			if !ok {
-				return 0, 0, errors.Errorf("requested non-existent TSN %v", tsn)
+				return nil, 0, errors.Errorf("requested non-existent TSN %v", tsn)
 			}
 
 			if !c.acked {
 				nBytesAcked := a.inflightQueue.markAsAcked(tsn)
 
-				// Report the number of bytes acknowledged to the stream who sent this DATA
-				// chunk.
-				if s, ok := a.streams[c.streamIdentifier]; ok {
-					a.lock.Unlock()
-					s.onBufferReleased(nBytesAcked)
-					a.lock.Lock()
+				// Sum the number of bytes acknowledged per stream
+				if amount, ok := bytesAckedPerStream[c.streamIdentifier]; ok {
+					bytesAckedPerStream[c.streamIdentifier] = amount + nBytesAcked
+				} else {
+					bytesAckedPerStream[c.streamIdentifier] = nBytesAcked
 				}
 
-				totalBytesAcked += nBytesAcked
-
 				a.log.Tracef("[%s] tsn=%d has been sacked", a.name, c.tsn)
 
 				if c.nSent == 1 {
@@ -1267,7 +1261,7 @@ func (a *Association) processSelectiveAck(d *chunkSelectiveAck) (int, uint32, er
 		}
 	}
 
-	return totalBytesAcked, htna, nil
+	return bytesAckedPerStream, htna, nil
 }
 
 // The caller should hold the lock.
@@ -1355,7 +1349,7 @@ func (a *Association) processFastRetransmission(cumTSNAckPoint, htna uint32, cum
 			if !ok {
 				return errors.Errorf("requested non-existent TSN %v", tsn)
 			}
-			if !c.acked && !c.abandoned && c.missIndicator < 3 {
+			if !c.acked && !c.abandoned() && c.missIndicator < 3 {
 				c.missIndicator++
 				if c.missIndicator == 3 {
 					if !a.inFastRecovery {
@@ -1412,11 +1406,36 @@ func (a *Association) handleSack(d *chunkSelectiveAck) error {
 	}
 
 	// Process selective ack
-	totalBytesAcked, htna, err := a.processSelectiveAck(d)
+	bytesAckedPerStream, htna, err := a.processSelectiveAck(d)
 	if err != nil {
 		return err
 	}
 
+	var totalBytesAcked int
+	for _, nBytesAcked := range bytesAckedPerStream {
+		totalBytesAcked += nBytesAcked
+	}
+
+	cumTSNAckPointAdvanced := false
+	if sna32LT(a.cumulativeTSNAckPoint, d.cumulativeTSNAck) {
+		a.log.Tracef("[%s] SACK: cumTSN advanced: %d -> %d",
+			a.name,
+			a.cumulativeTSNAckPoint,
+			d.cumulativeTSNAck)
+
+		a.cumulativeTSNAckPoint = d.cumulativeTSNAck
+		cumTSNAckPointAdvanced = true
+		a.onCumulativeTSNAckPointAdvanced(totalBytesAcked)
+	}
+
+	for si, nBytesAcked := range bytesAckedPerStream {
+		if s, ok := a.streams[si]; ok {
+			a.lock.Unlock()
+			s.onBufferReleased(nBytesAcked)
+			a.lock.Lock()
+		}
+	}
+
 	// New rwnd value
 	// RFC 4960 sec 6.2.1.  Processing a Received SACK
 	// D)
@@ -1432,18 +1451,6 @@ func (a *Association) handleSack(d *chunkSelectiveAck) error {
 		a.rwnd = d.advertisedReceiverWindowCredit - bytesOutstanding
 	}
 
-	cumTSNAckPointAdvanced := false
-	if sna32LT(a.cumulativeTSNAckPoint, d.cumulativeTSNAck) {
-		a.log.Tracef("[%s] SACK: cumTSN advanced: %d -> %d",
-			a.name,
-			a.cumulativeTSNAckPoint,
-			d.cumulativeTSNAck)
-
-		a.cumulativeTSNAckPoint = d.cumulativeTSNAck
-		cumTSNAckPointAdvanced = true
-		a.onCumulativeTSNAckPointAdvanced(totalBytesAcked)
-	}
-
 	err = a.processFastRetransmission(d.cumulativeTSNAck, htna, cumTSNAckPointAdvanced)
 	if err != nil {
 		return err
@@ -1461,14 +1468,16 @@ func (a *Association) handleSack(d *chunkSelectiveAck) error {
 			if !ok {
 				break
 			}
-			if !c.abandoned {
+			if !c.abandoned() {
 				break
 			}
 			a.advancedPeerTSNAckPoint = i
 		}
 
 		// RFC 3758 Sec 3.5 C3
-		a.willSendForwardTSN = true
+		if sna32GT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
+			a.willSendForwardTSN = true
+		}
 		a.awakeWriteLoop()
 	}
 
@@ -1496,9 +1505,6 @@ func (a *Association) createForwardTSN() *chunkForwardTSN {
 		if !ok {
 			break
 		}
-		if c.acked {
-			continue
-		}
 
 		ssn, ok := streamMap[c.streamIdentifier]
 		if !ok {
@@ -1524,7 +1530,7 @@ func (a *Association) createForwardTSN() *chunkForwardTSN {
 			sequence:   ssn,
 		})
 	}
-	a.log.Tracef("[%s] building fwdtsn: newCumulativeTSN=%d - %s", a.name, fwdtsn.newCumulativeTSN, streamStr)
+	a.log.Tracef("[%s] building fwdtsn: newCumulativeTSN=%d cumTSN=%d - %s", a.name, fwdtsn.newCumulativeTSN, a.cumulativeTSNAckPoint, streamStr)
 
 	return fwdtsn
 }
@@ -1620,10 +1626,19 @@ func (a *Association) handleForwardTSN(c *chunkForwardTSN) []*packet {
 	// from the reassemblyQueue.
 	for _, forwarded := range c.streams {
 		if s, ok := a.streams[forwarded.identifier]; ok {
-			s.handleForwardTSN(c.newCumulativeTSN, forwarded.sequence)
+			s.handleForwardTSNForOrdered(c.newCumulativeTSN, forwarded.sequence)
 		}
 	}
 
+	// TSN may be forewared for unordered chunks. ForwardTSN chunk does not
+	// report which stream identifier it skipped for unordered chunks.
+	// Therefore, we need to broadcast this event to all existing streams for
+	// unordered chunks.
+	// See https://github.com/pion/sctp/issues/106
+	for _, s := range a.streams {
+		s.handleForwardTSNForUnordered(c.newCumulativeTSN)
+	}
+
 	return a.handlePeerLastTSNAndAcknowledgement(false)
 }
 
@@ -1709,6 +1724,11 @@ func (a *Association) movePendingDataChunkToInflightQueue(c *chunkPayloadData) {
 		a.log.Errorf("[%s] failed to pop from pending queue: %s", a.name, err.Error())
 	}
 
+	// Mark all fragements are in-flight now
+	if c.endingFragment {
+		c.setAllInflight()
+	}
+
 	// Assign TSN
 	c.tsn = a.generateNextTSN()
 
@@ -1717,8 +1737,8 @@ func (a *Association) movePendingDataChunkToInflightQueue(c *chunkPayloadData) {
 
 	a.checkPartialReliabilityStatus(c)
 
-	a.log.Tracef("[%s] sending tsn=%d ssn=%d sent=%d len=%d",
-		a.name, c.tsn, c.streamSequenceNumber, c.nSent, len(c.userData))
+	a.log.Tracef("[%s] sending tsn=%d ssn=%d sent=%d len=%d (%v,%v)",
+		a.name, c.tsn, c.streamSequenceNumber, c.nSent, len(c.userData), c.beginningFragment, c.endingFragment)
 
 	// Push it into the inflightQueue
 	a.inflightQueue.pushNoCheck(c)
@@ -1847,14 +1867,14 @@ func (a *Association) checkPartialReliabilityStatus(c *chunkPayloadData) {
 		s.lock.RLock()
 		if s.reliabilityType == ReliabilityTypeRexmit {
 			if c.nSent >= s.reliabilityValue {
-				c.abandoned = true
-				a.log.Tracef("[%s] final (abandoned) tsn=%d (remix: %d)", a.name, c.tsn, c.nSent)
+				c.setAbandoned(true)
+				a.log.Tracef("[%s] marked as abandoned: tsn=%d (remix: %d)", a.name, c.tsn, c.nSent)
 			}
 		} else if s.reliabilityType == ReliabilityTypeTimed {
 			elapsed := int64(time.Since(c.since).Seconds() * 1000)
 			if elapsed >= int64(s.reliabilityValue) {
-				c.abandoned = true
-				a.log.Tracef("[%s] final (abandoned) tsn=%d (timed: %d)", a.name, c.tsn, elapsed)
+				c.setAbandoned(true)
+				a.log.Tracef("[%s] marked as abandoned: tsn=%d (timed: %d)", a.name, c.tsn, elapsed)
 			}
 		}
 		s.lock.RUnlock()
@@ -2067,8 +2087,43 @@ func (a *Association) onRetransmissionTimeout(id int, nRtos uint) {
 		a.log.Tracef("[%s] updated cwnd=%d ssthresh=%d inflight=%d (RTO)",
 			a.name, a.cwnd, a.ssthresh, a.inflightQueue.getNumBytes())
 
+		// RFC 3758 sec 3.5
+		//  A5) Any time the T3-rtx timer expires, on any destination, the sender
+		//  SHOULD try to advance the "Advanced.Peer.Ack.Point" by following
+		//  the procedures outlined in C2 - C5.
+		if a.useForwardTSN {
+			// RFC 3758 Sec 3.5 C2
+			for i := a.advancedPeerTSNAckPoint + 1; ; i++ {
+				c, ok := a.inflightQueue.get(i)
+				if !ok {
+					break
+				}
+				if !c.abandoned() {
+					break
+				}
+				a.advancedPeerTSNAckPoint = i
+			}
+
+			// RFC 3758 Sec 3.5 C3
+			if sna32GT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
+				a.willSendForwardTSN = true
+			}
+		}
+
 		a.log.Debugf("[%s] T3-rtx timed out: nRtos=%d cwnd=%d ssthresh=%d", a.name, nRtos, a.cwnd, a.ssthresh)
 
+		/*
+			a.log.Debugf("   - advancedPeerTSNAckPoint=%d", a.advancedPeerTSNAckPoint)
+			a.log.Debugf("   - cumulativeTSNAckPoint=%d", a.cumulativeTSNAckPoint)
+			a.inflightQueue.updateSortedKeys()
+			for i, tsn := range a.inflightQueue.sorted {
+				if c, ok := a.inflightQueue.get(tsn); ok {
+					a.log.Debugf("   - [%d] tsn=%d acked=%v abandoned=%v (%v,%v) len=%d",
+						i, c.tsn, c.acked, c.abandoned(), c.beginningFragment, c.endingFragment, len(c.userData))
+				}
+			}
+		*/
+
 		a.inflightQueue.markAllToRetrasmit()
 		a.awakeWriteLoop()
 
diff --git a/chunk_payload_data.go b/chunk_payload_data.go
index 01c5102..3e5e346 100644
--- a/chunk_payload_data.go
+++ b/chunk_payload_data.go
@@ -61,13 +61,16 @@ type chunkPayloadData struct {
 	missIndicator uint32
 
 	// Partial-reliability parameters used only by sender
-	since     time.Time
-	nSent     uint32 // number of transmission made for this chunk
-	abandoned bool
+	since        time.Time
+	nSent        uint32 // number of transmission made for this chunk
+	_abandoned   bool
+	_allInflight bool // valid only with the first fragment
 
 	// Retransmission flag set when T1-RTX timeout occurred and this
 	// chunk is still in the inflight queue
 	retransmit bool
+
+	head *chunkPayloadData // link to the head of the fragment
 }
 
 const (
@@ -165,3 +168,28 @@ func (p *chunkPayloadData) check() (abort bool, err error) {
 func (p *chunkPayloadData) String() string {
 	return fmt.Sprintf("%s\n%d", p.chunkHeader, p.tsn)
 }
+
+func (p *chunkPayloadData) abandoned() bool {
+	if p.head != nil {
+		return p.head._abandoned && p.head._allInflight
+	}
+	return p._abandoned && p._allInflight
+}
+
+func (p *chunkPayloadData) setAbandoned(abandoned bool) {
+	if p.head != nil {
+		p.head._abandoned = abandoned
+		return
+	}
+	p._abandoned = abandoned
+}
+
+func (p *chunkPayloadData) setAllInflight() {
+	if p.endingFragment {
+		if p.head != nil {
+			p.head._allInflight = true
+		} else {
+			p._allInflight = true
+		}
+	}
+}
diff --git a/payload_queue.go b/payload_queue.go
index 1877d57..8bf4f5f 100644
--- a/payload_queue.go
+++ b/payload_queue.go
@@ -162,7 +162,7 @@ func (q *payloadQueue) getLastTSNReceived() (uint32, bool) {
 
 func (q *payloadQueue) markAllToRetrasmit() {
 	for _, c := range q.chunkMap {
-		if c.acked || c.abandoned {
+		if c.acked || c.abandoned() {
 			continue
 		}
 		c.retransmit = true
diff --git a/reassembly_queue.go b/reassembly_queue.go
index 54e2399..11456a2 100644
--- a/reassembly_queue.go
+++ b/reassembly_queue.go
@@ -290,51 +290,48 @@ func (r *reassemblyQueue) read(buf []byte) (int, PayloadProtocolIdentifier, erro
 	return nWritten, ppi, err
 }
 
-func (r *reassemblyQueue) forwardTSN(newCumulativeTSN uint32, unordered bool, lastSSN uint16) {
-	if unordered {
-		// This stream is configured to use unordered data. (The given SSN
-		// has no significance)
-		// Remove all fragments in the unordered sets that contains chunks
-		// equal to or older than `newCumulativeTSN`.
-		// We know all sets in the r.unordered are complete ones.
-		// Just remove chunks that are equal to or older than newCumulativeTSN
-		// from the unorderedChunks
-		lastIdx := -1
-		for i, c := range r.unorderedChunks {
-			if sna32GT(c.tsn, newCumulativeTSN) {
-				break
-			}
-			lastIdx = i
-		}
-		if lastIdx >= 0 {
-			for _, c := range r.unorderedChunks[0 : lastIdx+1] {
-				r.subtractNumBytes(len(c.userData))
-			}
-			r.unorderedChunks = r.unorderedChunks[lastIdx+1:]
-		}
-	} else {
-		// This stream is configured to use ordered data.
-		// Use lastSSN to locate a chunkSet then remove it if the set has
-		// not been complete
-		keep := []*chunkSet{}
-		for _, set := range r.ordered {
-			if sna16LTE(set.ssn, lastSSN) {
-				if !set.isComplete() {
-					// drop the set
-					for _, c := range set.chunks {
-						r.subtractNumBytes(len(c.userData))
-					}
-					continue
+func (r *reassemblyQueue) forwardTSNForOrdered(newCumulativeTSN uint32, lastSSN uint16) {
+	// Use lastSSN to locate a chunkSet then remove it if the set has
+	// not been complete
+	keep := []*chunkSet{}
+	for _, set := range r.ordered {
+		if sna16LTE(set.ssn, lastSSN) {
+			if !set.isComplete() {
+				// drop the set
+				for _, c := range set.chunks {
+					r.subtractNumBytes(len(c.userData))
 				}
+				continue
 			}
-			keep = append(keep, set)
 		}
-		r.ordered = keep
+		keep = append(keep, set)
+	}
+	r.ordered = keep
 
-		// Finally, forward nextSSN
-		if sna16LTE(r.nextSSN, lastSSN) {
-			r.nextSSN = lastSSN + 1
+	// Finally, forward nextSSN
+	if sna16LTE(r.nextSSN, lastSSN) {
+		r.nextSSN = lastSSN + 1
+	}
+}
+
+func (r *reassemblyQueue) forwardTSNForUnordered(newCumulativeTSN uint32) {
+	// Remove all fragments in the unordered sets that contains chunks
+	// equal to or older than `newCumulativeTSN`.
+	// We know all sets in the r.unordered are complete ones.
+	// Just remove chunks that are equal to or older than newCumulativeTSN
+	// from the unorderedChunks
+	lastIdx := -1
+	for i, c := range r.unorderedChunks {
+		if sna32GT(c.tsn, newCumulativeTSN) {
+			break
+		}
+		lastIdx = i
+	}
+	if lastIdx >= 0 {
+		for _, c := range r.unorderedChunks[0 : lastIdx+1] {
+			r.subtractNumBytes(len(c.userData))
 		}
+		r.unorderedChunks = r.unorderedChunks[lastIdx+1:]
 	}
 }
 
diff --git a/stream.go b/stream.go
index 87025cc..1793381 100644
--- a/stream.go
+++ b/stream.go
@@ -124,14 +124,45 @@ func (s *Stream) handleData(pd *chunkPayloadData) {
 	}
 }
 
-func (s *Stream) handleForwardTSN(newCumulativeTSN uint32, ssn uint16) {
+func (s *Stream) handleForwardTSNForOrdered(newCumulativeTSN uint32, ssn uint16) {
 	var readable bool
-	s.lock.Lock()
-	// Remove all chunks older than or equal to the new TSN from
-	// the reassemblyQueue.
-	s.reassemblyQueue.forwardTSN(newCumulativeTSN, s.unordered, ssn)
-	readable = s.reassemblyQueue.isReadable()
-	s.lock.Unlock()
+
+	func() {
+		s.lock.Lock()
+		defer s.lock.Unlock()
+
+		if s.unordered {
+			return // unordered chunks are handled by handleForwardUnordered method
+		}
+
+		// Remove all chunks older than or equal to the new TSN from
+		// the reassemblyQueue.
+		s.reassemblyQueue.forwardTSNForOrdered(newCumulativeTSN, ssn)
+		readable = s.reassemblyQueue.isReadable()
+	}()
+
+	// Notify the reader asynchronously if there's a data chunk to read.
+	if readable {
+		s.readNotifier.Signal()
+	}
+}
+
+func (s *Stream) handleForwardTSNForUnordered(newCumulativeTSN uint32) {
+	var readable bool
+
+	func() {
+		s.lock.Lock()
+		defer s.lock.Unlock()
+
+		if !s.unordered {
+			return // ordered chunks are handled by handleForwardTSNOrdered method
+		}
+
+		// Remove all chunks older than or equal to the new TSN from
+		// the reassemblyQueue.
+		s.reassemblyQueue.forwardTSNForUnordered(newCumulativeTSN)
+		readable = s.reassemblyQueue.isReadable()
+	}()
 
 	// Notify the reader asynchronously if there's a data chunk to read.
 	if readable {
@@ -175,6 +206,7 @@ func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPa
 	unordered := ppi != PayloadTypeWebRTCDCEP && s.unordered
 
 	var chunks []*chunkPayloadData
+	var head *chunkPayloadData
 	for remaining != 0 {
 		fragmentSize := min32(s.association.maxPayloadSize, remaining)
 
@@ -192,6 +224,11 @@ func (s *Stream) packetize(raw []byte, ppi PayloadProtocolIdentifier) []*chunkPa
 			immediateSack:        false,
 			payloadType:          ppi,
 			streamSequenceNumber: s.sequenceNumber,
+			head:                 head,
+		}
+
+		if head == nil {
+			head = chunk
 		}
 
 		chunks = append(chunks, chunk)


Test patch:
diff --git a/association_test.go b/association_test.go
index 4778203..18524b6 100644
--- a/association_test.go
+++ b/association_test.go
@@ -1096,7 +1096,7 @@ func TestCreateForwardTSN(t *testing.T) {
 			streamSequenceNumber: 2,
 			userData:             []byte("ABC"),
 			nSent:                1,
-			abandoned:            true,
+			_abandoned:           true,
 		})
 
 		fwdtsn := a.createForwardTSN()
@@ -1123,7 +1123,7 @@ func TestCreateForwardTSN(t *testing.T) {
 			streamSequenceNumber: 2,
 			userData:             []byte("ABC"),
 			nSent:                1,
-			abandoned:            true,
+			_abandoned:           true,
 		})
 		a.inflightQueue.pushNoCheck(&chunkPayloadData{
 			beginningFragment:    true,
@@ -1133,7 +1133,7 @@ func TestCreateForwardTSN(t *testing.T) {
 			streamSequenceNumber: 3,
 			userData:             []byte("DEF"),
 			nSent:                1,
-			abandoned:            true,
+			_abandoned:           true,
 		})
 		a.inflightQueue.pushNoCheck(&chunkPayloadData{
 			beginningFragment:    true,
@@ -1143,7 +1143,7 @@ func TestCreateForwardTSN(t *testing.T) {
 			streamSequenceNumber: 1,
 			userData:             []byte("123"),
 			nSent:                1,
-			abandoned:            true,
+			_abandoned:           true,
 		})
 
 		fwdtsn := a.createForwardTSN()
diff --git a/reassembly_queue_test.go b/reassembly_queue_test.go
index c72357d..e12af1d 100644
--- a/reassembly_queue_test.go
+++ b/reassembly_queue_test.go
@@ -400,7 +400,7 @@ func TestReassemblyQueue(t *testing.T) {
 		assert.False(t, complete, "chunk set should not be complete yet")
 		assert.Equal(t, 9, rq.getNumBytes(), "num bytes mismatch")
 
-		rq.forwardTSN(13, false, ssnDropped)
+		rq.forwardTSNForOrdered(13, ssnDropped)
 
 		assert.Equal(t, 1, len(rq.ordered), "there should be one chunk left")
 		assert.Equal(t, 3, rq.getNumBytes(), "num bytes mismatch")
@@ -456,7 +456,7 @@ func TestReassemblyQueue(t *testing.T) {
 
 		// At this point, there are 3 chunks in the rq.unorderedChunks.
 		// This call should remove chunks with tsn equals to 13 or older.
-		rq.forwardTSN(13, true, ssnDropped)
+		rq.forwardTSNForUnordered(13)
 
 		// As a result, there should be one chunk (tsn=14)
 		assert.Equal(t, 1, len(rq.unorderedChunks), "there should be one chunk kept")