Repository: pion/sctp
Pull request #81 (2019-10-22 21:16:14)

Retransmit RECONFIG
See #45 

In addition to the retransmission of RECONFIG chunk, following bugs were found and also fixed:

* Reconfig was not handled in the case where TSN was forwarded
* The delayed ack handling resolved in #79 for DATA chunk was not applied to ForwardTSN chunk. (I have made a common routine for handleData() and handleForwardTSN())
* Added a test for RECONFIG retransmission
* Refactored a.gatherOutbound() for complexity error by golangci-lint 

codecov-io: # [Codecov](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=h1) Report
> Merging [#81](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=desc) into [master](https://codecov.io/gh/pion/sctp/commit/cd108564f452c810002faa11a60782585a29024b?src=pr&el=desc) will **increase** coverage by `0.41%`.
> The diff coverage is `81.65%`.

[![Impacted file tree graph](https://codecov.io/gh/pion/sctp/pull/81/graphs/tree.svg?width=650&token=AB8S3U6xR9&height=150&src=pr)](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=tree)

```diff
@@            Coverage Diff             @@
##           master      #81      +/-   ##
==========================================
+ Coverage   76.72%   77.13%   +0.41%     
==========================================
  Files          43       43              
  Lines        2690     2712      +22     
==========================================
+ Hits         2064     2092      +28     
+ Misses        502      494       -8     
- Partials      124      126       +2
```


| [Impacted Files](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [paramheader.go](https://codecov.io/gh/pion/sctp/pull/81/diff?src=pr&el=tree#diff-cGFyYW1oZWFkZXIuZ28=) | `92.3% <100%> (+7.69%)` | :arrow_up: |
| [association.go](https://codecov.io/gh/pion/sctp/pull/81/diff?src=pr&el=tree#diff-YXNzb2NpYXRpb24uZ28=) | `82.71% <81.48%> (+0.26%)` | :arrow_up: |
| [paramtype.go](https://codecov.io/gh/pion/sctp/pull/81/diff?src=pr&el=tree#diff-cGFyYW10eXBlLmdv) | `16.66% <0%> (+3.33%)` | :arrow_up: |
| [chunk\_reconfig.go](https://codecov.io/gh/pion/sctp/pull/81/diff?src=pr&el=tree#diff-Y2h1bmtfcmVjb25maWcuZ28=) | `80.48% <0%> (+7.31%)` | :arrow_up: |

------

[Continue to review full report at Codecov](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=footer). Last update [cd10856...c154f05](https://codecov.io/gh/pion/sctp/pull/81?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).

enobufs: Thanks, Sean!
No worries, I have been making sure to implement tests as well. This one was the last big essential feature we have been missing!

Patch:
diff --git a/association.go b/association.go
index 6241292..e579f2c 100644
--- a/association.go
+++ b/association.go
@@ -40,6 +40,7 @@ const (
 	timerT1Init int = iota
 	timerT1Cookie
 	timerT3RTX
+	timerReconfig
 )
 
 // ack mode (for testing)
@@ -109,14 +110,15 @@ type Association struct {
 
 	netConn net.Conn
 
-	peerVerificationTag uint32
-	myVerificationTag   uint32
-	state               uint32
-	myNextTSN           uint32 // nextTSN
-	peerLastTSN         uint32 // lastRcvdTSN
-	minTSN2MeasureRTT   uint32 // for RTT measurement
-	willSendForwardTSN  bool
-	willRetransmitFast  bool
+	peerVerificationTag    uint32
+	myVerificationTag      uint32
+	state                  uint32
+	myNextTSN              uint32 // nextTSN
+	peerLastTSN            uint32 // lastRcvdTSN
+	minTSN2MeasureRTT      uint32 // for RTT measurement
+	willSendForwardTSN     bool
+	willRetransmitFast     bool
+	willRetransmitReconfig bool
 
 	// Reconfig
 	myNextRSN        uint32
@@ -149,11 +151,12 @@ type Association struct {
 	fastRecoverExitPoint uint32
 
 	// RTX & Ack timer
-	rtoMgr   *rtoManager
-	t1Init   *rtxTimer
-	t1Cookie *rtxTimer
-	t3RTX    *rtxTimer
-	ackTimer *ackTimer
+	rtoMgr    *rtoManager
+	t1Init    *rtxTimer
+	t1Cookie  *rtxTimer
+	t3RTX     *rtxTimer
+	tReconfig *rtxTimer
+	ackTimer  *ackTimer
 
 	// Chunks stored for retransmission
 	storedInit       *chunkInit
@@ -281,7 +284,8 @@ func createAssociation(config Config) *Association {
 
 	a.t1Init = newRTXTimer(timerT1Init, a, maxInitRetrans)
 	a.t1Cookie = newRTXTimer(timerT1Cookie, a, maxInitRetrans)
-	a.t3RTX = newRTXTimer(timerT3RTX, a, noMaxRetrans) // retransmit forever
+	a.t3RTX = newRTXTimer(timerT3RTX, a, noMaxRetrans)        // retransmit forever
+	a.tReconfig = newRTXTimer(timerReconfig, a, noMaxRetrans) // retransmit forever
 	a.ackTimer = newAckTimer(a)
 
 	return a
@@ -387,6 +391,7 @@ func (a *Association) closeAllTimers() {
 	a.t1Init.close()
 	a.t1Cookie.close()
 	a.t3RTX.close()
+	a.tReconfig.close()
 	a.ackTimer.close()
 }
 
@@ -503,50 +508,46 @@ func (a *Association) handleInbound(raw []byte) error {
 	return nil
 }
 
-// handleOutbound handles outgoing processes
-func (a *Association) gatherOutbound() [][]byte {
-	a.lock.Lock()
-	defer a.lock.Unlock()
-
-	rawPackets := [][]byte{}
-
-	if a.controlQueue.size() > 0 {
-		for _, p := range a.controlQueue.popAll() {
-			raw, err := p.marshal()
-			if err != nil {
-				a.log.Warnf("[%s] failed to serialize a control packet", a.name)
-				continue
-			}
-			rawPackets = append(rawPackets, raw)
+// The caller should hold the lock
+func (a *Association) gatherOutboundDataAndReconfigPackets(rawPackets [][]byte) [][]byte {
+	for _, p := range a.getDataPacketsToRetransmit() {
+		raw, err := p.marshal()
+		if err != nil {
+			a.log.Warnf("[%s] failed to serialize a DATA packet to be retransmitted", a.name)
+			continue
 		}
+		rawPackets = append(rawPackets, raw)
 	}
 
-	state := a.getState()
-
-	if state == established {
-		for _, p := range a.getDataPacketsToRetransmit() {
+	// Pop unsent data chunks from the pending queue to send as much as
+	// cwnd and rwnd allow.
+	chunks, sisToReset := a.popPendingDataChunksToSend()
+	if len(chunks) > 0 {
+		// Start timer. (noop if already started)
+		a.log.Tracef("[%s] T3-rtx timer start (pt1)", a.name)
+		a.t3RTX.start(a.rtoMgr.getRTO())
+		for _, p := range a.bundleDataChunksIntoPackets(chunks) {
 			raw, err := p.marshal()
 			if err != nil {
-				a.log.Warnf("[%s] failed to serialize a DATA packet to be retransmitted", a.name)
+				a.log.Warnf("[%s] failed to serialize a DATA packet", a.name)
 				continue
 			}
 			rawPackets = append(rawPackets, raw)
 		}
+	}
 
-		// Pop unsent data chunks from the pending queue to send as much as
-		// cwnd and rwnd allow.
-		chunks, sisToReset := a.popPendingDataChunksToSend()
-		if len(chunks) > 0 {
-			// Start timer. (noop if already started)
-			a.log.Tracef("[%s] T3-rtx timer start (pt1)", a.name)
-			a.t3RTX.start(a.rtoMgr.getRTO())
-			for _, p := range a.bundleDataChunksIntoPackets(chunks) {
+	if len(sisToReset) > 0 || a.willRetransmitReconfig {
+		if a.willRetransmitReconfig {
+			a.willRetransmitReconfig = false
+			a.log.Debugf("[%s] retransmit %d RECONFIG chunk(s)", a.name, len(a.reconfigs))
+			for _, c := range a.reconfigs {
+				p := a.createPacket([]chunk{c})
 				raw, err := p.marshal()
 				if err != nil {
-					a.log.Warnf("[%s] failed to serialize a DATA packet", a.name)
-					continue
+					a.log.Warnf("[%s] failed to serialize a RECONFIG packet to be retransmitted", a.name)
+				} else {
+					rawPackets = append(rawPackets, raw)
 				}
-				rawPackets = append(rawPackets, raw)
 			}
 		}
 
@@ -563,92 +564,110 @@ func (a *Association) gatherOutbound() [][]byte {
 			a.reconfigs[rsn] = c // store in the map for retransmission
 			a.log.Debugf("[%s] sending RECONFIG: rsn=%d tsn=%d streams=%v",
 				a.name, rsn, a.myNextTSN-1, sisToReset)
-
 			p := a.createPacket([]chunk{c})
 			raw, err := p.marshal()
 			if err != nil {
-				a.log.Warnf("[%s] failed to serialize a RECONFIG packet to be retransmitted", a.name)
+				a.log.Warnf("[%s] failed to serialize a RECONFIG packet to be transmitted", a.name)
 			} else {
 				rawPackets = append(rawPackets, raw)
 			}
 		}
 
-		if a.willRetransmitFast {
-			a.willRetransmitFast = false
+		if len(a.reconfigs) > 0 {
+			a.tReconfig.start(a.rtoMgr.getRTO())
+		}
+	}
 
-			toFastRetrans := []chunk{}
-			fastRetransSize := commonHeaderSize
+	return rawPackets
+}
 
-			for i := 0; ; i++ {
-				c, ok := a.inflightQueue.get(a.cumulativeTSNAckPoint + uint32(i) + 1)
-				if !ok {
-					break // end of pending data
-				}
+// The caller should hold the lock
+func (a *Association) gatherOutboundFrastRetransmissionPackets(rawPackets [][]byte) [][]byte {
+	if a.willRetransmitFast {
+		a.willRetransmitFast = false
 
-				if c.acked || c.abandoned {
-					continue
-				}
+		toFastRetrans := []chunk{}
+		fastRetransSize := commonHeaderSize
 
-				if c.nSent > 1 || c.missIndicator < 3 {
-					continue
-				}
+		for i := 0; ; i++ {
+			c, ok := a.inflightQueue.get(a.cumulativeTSNAckPoint + uint32(i) + 1)
+			if !ok {
+				break // end of pending data
+			}
 
-				// RFC 4960 Sec 7.2.4 Fast Retransmit on Gap Reports
-				//  3)  Determine how many of the earliest (i.e., lowest TSN) DATA chunks
-				//      marked for retransmission will fit into a single packet, subject
-				//      to constraint of the path MTU of the destination transport
-				//      address to which the packet is being sent.  Call this value K.
-				//      Retransmit those K DATA chunks in a single packet.  When a Fast
-				//      Retransmit is being performed, the sender SHOULD ignore the value
-				//      of cwnd and SHOULD NOT delay retransmission for this single
-				//		packet.
-
-				dataChunkSize := dataChunkHeaderSize + uint32(len(c.userData))
-				if a.mtu < fastRetransSize+dataChunkSize {
-					break
-				}
+			if c.acked || c.abandoned {
+				continue
+			}
 
-				fastRetransSize += dataChunkSize
-				a.stats.incFastRetrans()
-				c.nSent++
-				a.checkPartialReliabilityStatus(c)
-				toFastRetrans = append(toFastRetrans, c)
-				a.log.Tracef("[%s] fast-retransmit: tsn=%d sent=%d htna=%d",
-					a.name, c.tsn, c.nSent, a.fastRecoverExitPoint)
+			if c.nSent > 1 || c.missIndicator < 3 {
+				continue
 			}
 
-			if len(toFastRetrans) > 0 {
-				raw, err := a.createPacket(toFastRetrans).marshal()
-				if err != nil {
-					a.log.Warnf("[%s] failed to serialize a DATA packet to be fast-retransmitted", a.name)
-				} else {
-					rawPackets = append(rawPackets, raw)
-				}
+			// RFC 4960 Sec 7.2.4 Fast Retransmit on Gap Reports
+			//  3)  Determine how many of the earliest (i.e., lowest TSN) DATA chunks
+			//      marked for retransmission will fit into a single packet, subject
+			//      to constraint of the path MTU of the destination transport
+			//      address to which the packet is being sent.  Call this value K.
+			//      Retransmit those K DATA chunks in a single packet.  When a Fast
+			//      Retransmit is being performed, the sender SHOULD ignore the value
+			//      of cwnd and SHOULD NOT delay retransmission for this single
+			//		packet.
+
+			dataChunkSize := dataChunkHeaderSize + uint32(len(c.userData))
+			if a.mtu < fastRetransSize+dataChunkSize {
+				break
 			}
+
+			fastRetransSize += dataChunkSize
+			a.stats.incFastRetrans()
+			c.nSent++
+			a.checkPartialReliabilityStatus(c)
+			toFastRetrans = append(toFastRetrans, c)
+			a.log.Tracef("[%s] fast-retransmit: tsn=%d sent=%d htna=%d",
+				a.name, c.tsn, c.nSent, a.fastRecoverExitPoint)
 		}
 
-		if a.ackState == ackStateImmediate {
-			a.ackState = ackStateIdle
-			sack := a.createSelectiveAckChunk()
-			a.log.Debugf("[%s] sending SACK: %s", a.name, sack.String())
-			raw, err := a.createPacket([]chunk{sack}).marshal()
+		if len(toFastRetrans) > 0 {
+			raw, err := a.createPacket(toFastRetrans).marshal()
 			if err != nil {
-				a.log.Warnf("[%s] failed to serialize a SACK packet", a.name)
+				a.log.Warnf("[%s] failed to serialize a DATA packet to be fast-retransmitted", a.name)
 			} else {
 				rawPackets = append(rawPackets, raw)
 			}
 		}
+	}
 
-		if a.willSendForwardTSN {
-			a.willSendForwardTSN = false
-			if sna32GT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
-				fwdtsn := a.createForwardTSN()
-				raw, err := a.createPacket([]chunk{fwdtsn}).marshal()
-				if err != nil {
-					a.log.Warnf("[%s] failed to serialize a Forward TSN packet", a.name)
-				} else {
-					rawPackets = append(rawPackets, raw)
-				}
+	return rawPackets
+}
+
+// The caller should hold the lock
+func (a *Association) gatherOutboundSackPackets(rawPackets [][]byte) [][]byte {
+	if a.ackState == ackStateImmediate {
+		a.ackState = ackStateIdle
+		sack := a.createSelectiveAckChunk()
+		a.log.Debugf("[%s] sending SACK: %s", a.name, sack.String())
+		raw, err := a.createPacket([]chunk{sack}).marshal()
+		if err != nil {
+			a.log.Warnf("[%s] failed to serialize a SACK packet", a.name)
+		} else {
+			rawPackets = append(rawPackets, raw)
+		}
+	}
+
+	return rawPackets
+}
+
+// The caller should hold the lock
+func (a *Association) gatherOutboundForwardTSNPackets(rawPackets [][]byte) [][]byte {
+	if a.willSendForwardTSN {
+		a.willSendForwardTSN = false
+		if sna32GT(a.advancedPeerTSNAckPoint, a.cumulativeTSNAckPoint) {
+			fwdtsn := a.createForwardTSN()
+			raw, err := a.createPacket([]chunk{fwdtsn}).marshal()
+			if err != nil {
+				a.log.Warnf("[%s] failed to serialize a Forward TSN packet", a.name)
+			} else {
+				rawPackets = append(rawPackets, raw)
 			}
 		}
 	}
@@ -656,6 +675,36 @@ func (a *Association) gatherOutbound() [][]byte {
 	return rawPackets
 }
 
+// gatherOutbound gathers outgoing packets
+func (a *Association) gatherOutbound() [][]byte {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+
+	rawPackets := [][]byte{}
+
+	if a.controlQueue.size() > 0 {
+		for _, p := range a.controlQueue.popAll() {
+			raw, err := p.marshal()
+			if err != nil {
+				a.log.Warnf("[%s] failed to serialize a control packet", a.name)
+				continue
+			}
+			rawPackets = append(rawPackets, raw)
+		}
+	}
+
+	state := a.getState()
+
+	if state == established {
+		rawPackets = a.gatherOutboundDataAndReconfigPackets(rawPackets)
+		rawPackets = a.gatherOutboundFrastRetransmissionPackets(rawPackets)
+		rawPackets = a.gatherOutboundSackPackets(rawPackets)
+		rawPackets = a.gatherOutboundForwardTSNPackets(rawPackets)
+	}
+
+	return rawPackets
+}
+
 func checkPacket(p *packet) error {
 	// All packets must adhere to these rules
 
@@ -991,9 +1040,21 @@ func (a *Association) handleData(d *chunkPayloadData) []*packet {
 		}
 	}
 
-	reply := []*packet{}
+	return a.handlePeerLastTSNAndAcknowledgement(d.immediateSack)
+}
+
+// A common routine for handleData and handleForwardTSN routines
+// The caller should hold the lock.
+func (a *Association) handlePeerLastTSNAndAcknowledgement(sackImmediately bool) []*packet {
+	var reply []*packet
 
 	// Try to advance peerLastTSN
+
+	// From RFC 3758 Sec 3.6:
+	//   .. and then MUST further advance its cumulative TSN point locally
+	//   if possible
+	// Meaning, if peerLastTSN+1 points to a chunk that is received,
+	// advance peerLastTSN until peerLastTSN+1 points to unreceived chunk.
 	for {
 		if _, popOk := a.payloadQueue.pop(a.peerLastTSN + 1); !popOk {
 			break
@@ -1001,7 +1062,7 @@ func (a *Association) handleData(d *chunkPayloadData) []*packet {
 		a.peerLastTSN++
 
 		for _, rstReq := range a.reconfigRequests {
-			resp := a.resetStreams(rstReq)
+			resp := a.resetStreamsIfAny(rstReq)
 			if resp != nil {
 				a.log.Debugf("[%s] RESET RESPONSE: %+v", a.name, resp)
 				reply = append(reply, resp)
@@ -1014,7 +1075,7 @@ func (a *Association) handleData(d *chunkPayloadData) []*packet {
 		a.log.Tracef("[%s] packetloss: %s", a.name, a.payloadQueue.getGapAckBlocksString(a.peerLastTSN))
 	}
 
-	if (a.ackState != ackStateImmediate && !d.immediateSack && !hasPacketLoss && a.ackMode == ackModeNormal) || a.ackMode == ackModeAlwaysDelay {
+	if (a.ackState != ackStateImmediate && !sackImmediately && !hasPacketLoss && a.ackMode == ackModeNormal) || a.ackMode == ackModeAlwaysDelay {
 		if a.ackState == ackStateIdle {
 			a.delayedAckTriggered = true
 		} else {
@@ -1552,18 +1613,6 @@ func (a *Association) handleForwardTSN(c *chunkForwardTSN) []*packet {
 		a.peerLastTSN++
 	}
 
-	// From RFC 3758 Sec 3.6:
-	//   .. and then MUST further advance its cumulative TSN point locally
-	//   if possible
-	// Meaning, if peerLastTSN+1 points to a chunk that is received,
-	// advance peerLastTSN until peerLastTSN+1 points to unreceived chunk.
-	for {
-		if _, popOk := a.payloadQueue.pop(a.peerLastTSN + 1); !popOk {
-			break
-		}
-		a.peerLastTSN++
-	}
-
 	// Report new peerLastTSN value and abandoned largest SSN value to
 	// corresponding streams so that the abandoned chunks can be removed
 	// from the reassemblyQueue.
@@ -1573,20 +1622,7 @@ func (a *Association) handleForwardTSN(c *chunkForwardTSN) []*packet {
 		}
 	}
 
-	hasPacketLoss := (a.payloadQueue.size() > 0)
-
-	if a.ackState != ackStateImmediate && !hasPacketLoss {
-		// Will send delayed ack in the next ack timeout
-		a.ackState = ackStateDelay
-		a.ackTimer.start()
-	} else {
-		// Send SACK now!
-		a.ackState = ackStateImmediate
-		a.ackTimer.stop()
-		a.awakeWriteLoop()
-	}
-
-	return nil
+	return a.handlePeerLastTSNAndAcknowledgement(false)
 }
 
 func (a *Association) sendResetRequest(streamIdentifier uint16) error {
@@ -1619,7 +1655,7 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 	switch p := raw.(type) {
 	case *paramOutgoingResetRequest:
 		a.reconfigRequests[p.reconfigRequestSequenceNumber] = p
-		resp := a.resetStreams(p)
+		resp := a.resetStreamsIfAny(p)
 		if resp != nil {
 			return resp, nil
 		}
@@ -1627,7 +1663,9 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 
 	case *paramReconfigResponse:
 		delete(a.reconfigs, p.reconfigResponseSequenceNumber)
-
+		if len(a.reconfigs) == 0 {
+			a.tReconfig.stop()
+		}
 		return nil, nil
 	default:
 		return nil, errors.Errorf("unexpected parameter type %T", p)
@@ -1635,7 +1673,7 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 }
 
 // The caller should hold the lock.
-func (a *Association) resetStreams(p *paramOutgoingResetRequest) *packet {
+func (a *Association) resetStreamsIfAny(p *paramOutgoingResetRequest) *packet {
 	result := reconfigResultSuccessPerformed
 	if sna32LTE(p.senderLastTSN, a.peerLastTSN) {
 		a.log.Debugf("[%s] resetStream(): senderLastTSN=%d <= peerLastTSN=%d",
@@ -2032,6 +2070,11 @@ func (a *Association) onRetransmissionTimeout(id int, nRtos uint) {
 
 		return
 	}
+
+	if id == timerReconfig {
+		a.willRetransmitReconfig = true
+		a.awakeWriteLoop()
+	}
 }
 
 func (a *Association) onRetransmissionFailure(id int) {
diff --git a/paramheader.go b/paramheader.go
index 9cf7d37..d88642b 100644
--- a/paramheader.go
+++ b/paramheader.go
@@ -2,6 +2,7 @@ package sctp
 
 import (
 	"encoding/binary"
+	"encoding/hex"
 	"fmt"
 
 	"github.com/pkg/errors"
@@ -58,5 +59,5 @@ func (p *paramHeader) length() int {
 
 // String makes paramHeader printable
 func (p paramHeader) String() string {
-	return fmt.Sprintf("%s (%d): %s", p.typ, p.len, p.raw)
+	return fmt.Sprintf("%s (%d): %s", p.typ, p.len, hex.Dump(p.raw))
 }


Test patch:
diff --git a/association_test.go b/association_test.go
index 646c4aa..ce29fb9 100644
--- a/association_test.go
+++ b/association_test.go
@@ -1186,10 +1186,12 @@ func TestHandleForwardTSN(t *testing.T) {
 		p := a.handleForwardTSN(fwdtsn)
 
 		a.lock.Lock()
-		ackState := a.ackState
+		delayedAckTriggered := a.delayedAckTriggered
+		immediateAckTriggered := a.immediateAckTriggered
 		a.lock.Unlock()
 		assert.Equal(t, a.peerLastTSN, prevTSN+3, "peerLastTSN should advance by 3 ")
-		assert.Equal(t, ackStateDelay, ackState, "delayed sack should be requested")
+		assert.True(t, delayedAckTriggered, "delayed sack should be triggered")
+		assert.False(t, immediateAckTriggered, "immediate sack should NOT be triggered")
 		assert.Nil(t, p, "should return nil")
 	})
 
@@ -1221,10 +1223,12 @@ func TestHandleForwardTSN(t *testing.T) {
 		p := a.handleForwardTSN(fwdtsn)
 
 		a.lock.Lock()
-		ackState := a.ackState
+		delayedAckTriggered := a.delayedAckTriggered
+		immediateAckTriggered := a.immediateAckTriggered
 		a.lock.Unlock()
 		assert.Equal(t, a.peerLastTSN, prevTSN+2, "peerLastTSN should advance by 3")
-		assert.Equal(t, ackStateDelay, ackState, "delayed sack should be requested")
+		assert.True(t, delayedAckTriggered, "delayed sack should be triggered")
+		assert.False(t, immediateAckTriggered, "immediate sack should NOT be triggered")
 		assert.Nil(t, p, "should return nil")
 	})
 
@@ -1256,10 +1260,10 @@ func TestHandleForwardTSN(t *testing.T) {
 		p := a.handleForwardTSN(fwdtsn)
 
 		a.lock.Lock()
-		ackState := a.ackState
+		immediateAckTriggered := a.immediateAckTriggered
 		a.lock.Unlock()
 		assert.Equal(t, a.peerLastTSN, prevTSN+1, "peerLastTSN should advance by 1")
-		assert.Equal(t, ackStateImmediate, ackState, "immediate sack should be requested")
+		assert.True(t, immediateAckTriggered, "immediate sack should be triggered")
 
 		assert.Nil(t, p, "should return nil")
 	})
diff --git a/vnet_test.go b/vnet_test.go
index f181fb1..e445a83 100644
--- a/vnet_test.go
+++ b/vnet_test.go
@@ -21,14 +21,19 @@ type vNetEnvConfig struct {
 }
 
 type vNetEnv struct {
-	wan       *vnet.Router
-	net0      *vnet.Net
-	net1      *vnet.Net
-	numToDrop int
+	wan               *vnet.Router
+	net0              *vnet.Net
+	net1              *vnet.Net
+	numToDropData     int
+	numToDropReconfig int
 }
 
 func (venv *vNetEnv) dropNextDataChunk(numToDrop int) {
-	venv.numToDrop = numToDrop
+	venv.numToDropData = numToDrop
+}
+
+func (venv *vNetEnv) dropNextReconfigChunk(numToDrop int) {
+	venv.numToDropReconfig = numToDrop
 }
 
 func buildVNetEnv(cfg *vNetEnvConfig) (*vNetEnv, error) {
@@ -53,32 +58,39 @@ func buildVNetEnv(cfg *vNetEnvConfig) (*vNetEnv, error) {
 		var tsn uint32
 
 		return func(c vnet.Chunk) bool {
-			var hasDataChunkToDrop bool
-			if venv.numToDrop > 0 {
-				p := &packet{}
-				if err2 := p.unmarshal(c.UserData()); err2 != nil {
-					panic(fmt.Errorf("unable to parse SCTP packet"))
-				}
+			var toDrop bool
+			p := &packet{}
+			if err2 := p.unmarshal(c.UserData()); err2 != nil {
+				panic(fmt.Errorf("unable to parse SCTP packet"))
+			}
 
-			loop:
-				for i := 0; i < len(p.chunks); i++ {
-					switch chunk := p.chunks[i].(type) {
-					case *chunkPayloadData:
+		loop:
+			for i := 0; i < len(p.chunks); i++ {
+				switch chunk := p.chunks[i].(type) {
+				case *chunkPayloadData:
+					if venv.numToDropData > 0 {
 						if !lockedOnTSN {
 							tsn = chunk.tsn
 							lockedOnTSN = true
 							log.Infof("Chunk filter: lock on TSN %d", tsn)
 						}
 						if chunk.tsn == tsn {
-							hasDataChunkToDrop = true
-							venv.numToDrop--
+							toDrop = true
+							venv.numToDropData--
 							log.Infof("Chunk filter:  drop TSN %d", tsn)
 							break loop
 						}
 					}
+				case *chunkReconfig:
+					if venv.numToDropReconfig > 0 {
+						toDrop = true
+						venv.numToDropReconfig--
+						log.Infof("Chunk filter:  drop RECONFIG %s", chunk.String())
+						break loop
+					}
 				}
 			}
-			return !hasDataChunkToDrop
+			return !toDrop
 		}
 	}
 	wan.AddChunkFilter(tsnAutoLockOnFilter())
@@ -347,7 +359,7 @@ func TestRwndFull(t *testing.T) {
 	})
 }
 
-func TestStreamClose(t *testing.T) {
+func testStreamClose(t *testing.T, dropReconfig bool) {
 	lim := test.TimeOut(time.Second * 5)
 	defer lim.Stop()
 
@@ -478,6 +490,10 @@ func TestStreamClose(t *testing.T) {
 		close(clientStreamReady)
 
 		<-clientStartClose
+
+		// drop next 1 RECONFIG chunk
+		venv.dropNextReconfigChunk(1)
+
 		err = stream.Close()
 		assert.NoError(t, err, "should succeed")
 
@@ -485,6 +501,19 @@ func TestStreamClose(t *testing.T) {
 		<-done
 
 		<-shutDownClient
+
+		// Check if RECONFIG was actually dropped
+		assert.Equal(t, 0, venv.numToDropReconfig, "should be zero")
+
+		// Sleep enough time for reconfig response to come back
+		time.Sleep(100 * time.Millisecond)
+
+		// Verify there's no more pending reconfig
+		assoc.lock.RLock()
+		pendingReconfigs := len(assoc.reconfigs)
+		assoc.lock.RUnlock()
+		assert.Equal(t, 0, pendingReconfigs, "should be zero")
+
 		log.Info("client closing")
 	}()
 
@@ -505,3 +534,13 @@ func TestStreamClose(t *testing.T) {
 	<-serverShutDown
 	log.Info("all done")
 }
+
+func TestStreamClose(t *testing.T) {
+	t.Run("Normal close", func(t *testing.T) {
+		testStreamClose(t, false)
+	})
+
+	t.Run("Drop reconfig packet", func(t *testing.T) {
+		testStreamClose(t, true)
+	})
+}