Notify when a stream reset completes

association.go

@@ -273,9 +273,11 @@ type Association struct {
 	abortSentCh        chan struct{}
 
 	// Reconfig
-	myNextRSN        uint32
-	reconfigs        map[uint32]*chunkReconfig
-	reconfigRequests map[uint32]*paramOutgoingResetRequest
+	myNextRSN                    uint32
+	reconfigs                    map[uint32]*chunkReconfig
+	reconfigRequests             map[uint32]*paramOutgoingResetRequest
+	onStreamResetCompleteHandler func(streamID uint16)
+	streamResetStates            map[uint16]streamResetState
 
 	// Non-RFC internal data
 	sourcePort              uint16
@@ -841,6 +843,7 @@ func createAssociationFromConfigWithTsn(cfg *Config, tsn uint32) *Association {
 		streams:                 map[uint16]*Stream{},
 		reconfigs:               map[uint32]*chunkReconfig{},
 		reconfigRequests:        map[uint32]*paramOutgoingResetRequest{},
+		streamResetStates:       map[uint16]streamResetState{},
 		acceptCh:                make(chan *Stream, acceptChSize),
 		readLoopCloseCh:         make(chan struct{}),
 		awakeWriteLoopCh:        make(chan struct{}, 1),
@@ -3720,7 +3723,7 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 			return nil, nil //nolint:nilnil
 		}
 		if par.result == reconfigResultSuccessPerformed {
-			a.resetOutgoingStreamSequenceNumbers(par.reconfigResponseSequenceNumber)
+			a.completeOutgoingStreamReset(par.reconfigResponseSequenceNumber)
 		}
 		delete(a.reconfigs, par.reconfigResponseSequenceNumber)
 		if len(a.reconfigs) == 0 {
@@ -3734,7 +3737,7 @@ func (a *Association) handleReconfigParam(raw param) (*packet, error) {
 }
 
 // The caller should hold the lock.
-func (a *Association) resetOutgoingStreamSequenceNumbers(reconfigRequestSequenceNumber uint32) {
+func (a *Association) completeOutgoingStreamReset(reconfigRequestSequenceNumber uint32) {
 	reconfig := a.reconfigs[reconfigRequestSequenceNumber]
 	if reconfig == nil {
 		return
@@ -3747,6 +3750,7 @@ func (a *Association) resetOutgoingStreamSequenceNumbers(reconfigRequestSequence
 		if s, ok := a.streams[id]; ok {
 			s.resetOutgoingStreamSequenceNumbers()
 		}
+		a.completeStreamResetDirection(id, streamResetOutbound)
 	}
 }
 
@@ -3766,6 +3770,7 @@ func (a *Association) resetStreamsIfAny(resetRequest *paramOutgoingResetRequest)
 			a.lock.Lock()
 			a.log.Debugf("[%s] deleting stream %d", a.name, id)
 			delete(a.streams, s.streamIdentifier)
+			a.completeStreamResetDirection(s.streamIdentifier, streamResetInbound)
 		}
 		delete(a.reconfigRequests, resetRequest.reconfigRequestSequenceNumber)
 	} else {
@@ -3782,6 +3787,45 @@ func (a *Association) resetStreamsIfAny(resetRequest *paramOutgoingResetRequest)
 	}})
 }
 
+type streamResetState uint8
+
+const (
+	streamResetInbound streamResetState = 1 << iota
+	streamResetOutbound
+	streamResetBoth = streamResetInbound | streamResetOutbound
+)
+
+// The caller should hold the association lock.
+func (a *Association) completeStreamResetDirection(streamID uint16, direction streamResetState) {
+	if a.streamResetStates == nil {
+		a.streamResetStates = map[uint16]streamResetState{}
+	}
+	state := a.streamResetStates[streamID] | direction
+	if state != streamResetBoth {
+		a.streamResetStates[streamID] = state
+
+		return
+	}
+	delete(a.streamResetStates, streamID)
+	handler := a.onStreamResetCompleteHandler
+	if handler == nil {
+		return
+	}
+
+	a.lock.Unlock()
+	handler(streamID)
+	a.lock.Lock()
+}
+
+// OnStreamResetComplete sets a handler invoked after both directions of a
+// stream reset complete and the stream has been removed from the association.
+// A stream identifier is safe to reuse once the handler has been called.
+func (a *Association) OnStreamResetComplete(f func(streamID uint16)) {
+	a.lock.Lock()
+	defer a.lock.Unlock()
+	a.onStreamResetCompleteHandler = f
+}
+
 // Move the chunk peeked with a.pendingQueue.peek() to the inflightQueue.
 // The caller should hold the lock.
 func (a *Association) movePendingDataChunkToInflightQueue(chunkPayload *chunkPayloadData) {

association_stream_reset_complete_test.go

@@ -0,0 +1,84 @@
+// SPDX-FileCopyrightText: 2026 The Pion community <https://pion.ly>
+// SPDX-License-Identifier: MIT
+
+//go:build !js
+
+package sctp
+
+import (
+	"testing"
+	"time"
+
+	"github.com/pion/transport/v5/test"
+	"github.com/stretchr/testify/require"
+)
+
+// Both peers are notified once, after both reset directions completed and the
+// stream is gone.
+func TestStreamResetCompleteNotifiesBothAssociations(t *testing.T) {
+	const streamID = uint16(5)
+	bridge := test.NewBridge()
+	client, server, err := createNewAssociationPair(bridge, ackModeNoDelay, 0)
+	require.NoError(t, err)
+	defer closeAssociationPair(bridge, client, server)
+	clientStream, serverStream, err := establishSessionPair(bridge, client, server, streamID)
+	require.NoError(t, err)
+
+	completed := func(assoc *Association) chan uint16 {
+		events := make(chan uint16, 2)
+		assoc.OnStreamResetComplete(func(id uint16) {
+			assoc.lock.RLock()
+			_, present := assoc.streams[id]
+			assoc.lock.RUnlock()
+			require.False(t, present, "stream %d still registered", id)
+			events <- id
+		})
+
+		return events
+	}
+	clientCompleted, serverCompleted := completed(client), completed(server)
+
+	// Only one direction is reset: neither side may report completion.
+	require.NoError(t, clientStream.Close())
+	flushBuffers(bridge, client, server)
+	require.Empty(t, serverCompleted, "server reported completion after one direction")
+	require.Empty(t, clientCompleted, "client reported completion after one direction")
+
+	require.NoError(t, serverStream.Close())
+	for _, side := range []struct {
+		name   string
+		events chan uint16
+	}{{"server", serverCompleted}, {"client", clientCompleted}} {
+		deadline := time.Now().Add(5 * time.Second)
+		for len(side.events) == 0 {
+			require.True(t, time.Now().Before(deadline), "%s was not notified", side.name)
+			bridge.Process()
+		}
+		require.Equal(t, streamID, <-side.events)
+	}
+
+	// The identifier is free again. Closing one direction of the next
+	// stream using it must not report completion right away.
+	reusedClient, err := client.OpenStream(streamID, PayloadTypeWebRTCBinary)
+	require.NoError(t, err)
+	_, err = reusedClient.WriteSCTP([]byte("reused"), PayloadTypeWebRTCBinary)
+	require.NoError(t, err)
+	flushBuffers(bridge, client, server)
+	_, err = server.AcceptStream()
+	require.NoError(t, err)
+
+	require.NoError(t, reusedClient.Close())
+	deadline := time.Now().Add(5 * time.Second)
+	for {
+		bridge.Process()
+		server.lock.RLock()
+		_, present := server.streams[streamID]
+		server.lock.RUnlock()
+		if !present {
+			break
+		}
+		require.True(t, time.Now().Before(deadline), "server never saw the reset of the reused stream")
+	}
+	require.Empty(t, serverCompleted, "server reported completion after one direction")
+	require.Empty(t, clientCompleted, "client reported completion after one direction")
+}