From 76e06ca3087a45b5bf041cf2a5f804b53a5779fd Mon Sep 17 00:00:00 2001 From: idy Date: Wed, 16 Sep 2026 23:10:22 +0800 Subject: [PATCH] Notify when a stream reset completes An application that reuses stream identifiers, such as WebRTC data channels, has no way to learn when a stream's reset has finished. Reusing the identifier earlier lets a late reset request or response for the old stream act on the new one. Add OnStreamResetComplete. The handler is called once per stream, after the stream has been removed and both the incoming reset from the peer and the acknowledgement of our own outgoing reset have arrived, so the identifier can safely be reused. Co-Authored-By: Claude Fable 5.1 --- association.go | 54 +++++++++++++-- association_stream_reset_complete_test.go | 84 +++++++++++++++++++++++ 2 files changed, 133 insertions(+), 5 deletions(-) create mode 100644 association_stream_reset_complete_test.go diff --git a/association.go b/association.go index b0c2ad3c..564cc3ea 100644 --- a/association.go +++ b/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) { diff --git a/association_stream_reset_complete_test.go b/association_stream_reset_complete_test.go new file mode 100644 index 00000000..440ad4ff --- /dev/null +++ b/association_stream_reset_complete_test.go @@ -0,0 +1,84 @@ +// SPDX-FileCopyrightText: 2026 The Pion community +// 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") +}