fix: hold signal messages until the ReconnectResponse goes out (#4827)

* fix: hold signal messages until the ReconnectResponse goes out

Clients take the ReconnectResponse as the first message on a resumed or
migrated in signal connection, anything ahead of it is dropped. Hold
messages back until it has been written.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: keep queued participant updates on a path that always flushes

Queue only while the participant is not ready, that queue is always
drained by the join or reconnect response. Log a dropped SDP, it leaves
the negotiation waiting until the state machine recovers it.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: flush queued updates whenever the connection opens

Queue participant updates while the handshake is pending again, and give
the signaller a hook that fires when the connection opens, on an explicit
open and on the handshake window expiring, so the queue always drains.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: drive the handshake window off a timer and read the gate atomically

The window ran only when something asked whether the handshake was
pending, so a connection with nothing else to send held its queue.
Reading the gate under the lock the flush takes closes the race where an
update queued just after a flush.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

* fix: ignore a handshake timeout from a wait that has ended

Stop cannot cancel a timeout that is already running, so tag each wait
and let a timeout act only on its own. Count opens atomically in the
test, the timer fires on its own goroutine.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

pkg/rtc/participant.go

@@ -261,6 +261,10 @@ type ParticipantImpl struct {
 	state        atomic.Value // livekit.ParticipantInfo_State
 	disconnected chan struct{}
 
+	// a migrating in participant resumes on a reconnect response, the client takes it
+	// only as the first message on the resumed signal connection
+	reconnectResponseSent atomic.Bool
+
 	grants      atomic.Pointer[auth.ClaimGrants]
 	isPublisher atomic.Bool
 
@@ -612,9 +616,10 @@ func (p *ParticipantImpl) IsReady() bool {
 	state := p.State()
 
 	// when migrating, there is no JoinResponse, state transitions from JOINING -> ACTIVE -> DISCONNECTED
-	// so JOINING is considered ready.
+	// so JOINING is considered ready. The ReconnectResponse takes the place of the JoinResponse
+	// as the message the resumed signal connection opens with, so readiness waits for it.
 	if p.params.Migration {
-		return state != livekit.ParticipantInfo_DISCONNECTED
+		return state != livekit.ParticipantInfo_DISCONNECTED && p.reconnectResponseSent.Load()
 	}
 
 	// when not migrating, there is a JoinResponse, state transitions from JOINING -> JOINED -> ACTIVE -> DISCONNECTED
@@ -2071,8 +2076,9 @@ func (p *ParticipantImpl) setupSignalling() {
 		Participant: p,
 	})
 	p.signaller = signalling.NewSignallerAsync(signalling.SignallerAsyncParams{
-		Logger:      p.params.Logger,
-		Participant: p,
+		Logger:            p.params.Logger,
+		Participant:       p,
+		OnHandshakeOpened: p.flushQueuedUpdates,
 	})
 }
 

pkg/rtc/participant_internal_test.go

@@ -783,6 +783,7 @@ type participantOpts struct {
 	publisher       bool
 	clientConf      *livekit.ClientConfiguration
 	clientInfo      *livekit.ClientInfo
+	migration       bool
 }
 
 func newParticipantForTestWithOpts(identity livekit.ParticipantIdentity, opts *participantOpts) *ParticipantImpl {
@@ -837,6 +838,7 @@ func newParticipantForTestWithOpts(identity livekit.ParticipantIdentity, opts *p
 		VersionGenerator:       utils.NewDefaultTimedVersionGenerator(),
 		ParticipantListener:    &typesfakes.FakeLocalParticipantListener{},
 		ParticipantHelper:      &typesfakes.FakeLocalParticipantHelper{},
+		Migration:              opts.migration,
 	})
 	p.isPublisher.Store(opts.publisher)
 	p.updateState(livekit.ParticipantInfo_ACTIVE)
@@ -895,3 +897,153 @@ func TestUnsequencedReliableDataBufferedWhileJoining(t *testing.T) {
 		require.Empty(t, p.reliableDataInfo.joiningUnsequencedMessages)
 	})
 }
+
+func TestMigratingInParticipantWaitsForReconnectResponse(t *testing.T) {
+	// a migrating in participant resumes on a ReconnectResponse, the client takes it only
+	// as the first message on the resumed signal connection, so everything else waits
+	newMigratingParticipant := func() (*ParticipantImpl, *routingfakes.FakeMessageSink) {
+		p := newParticipantForTestWithOpts("test", &participantOpts{
+			migration:       true,
+			protocolVersion: 17,
+			clientInfo:      &livekit.ClientInfo{Sdk: livekit.ClientInfo_JS, Version: "2.15.2"},
+		})
+		return p, p.params.Sink.(*routingfakes.FakeMessageSink)
+	}
+
+	t.Run("not ready until the response is sent", func(t *testing.T) {
+		p, sink := newMigratingParticipant()
+		require.False(t, p.IsReady())
+
+		require.NoError(t, p.HandleReconnectAndSendResponse(
+			livekit.ReconnectReason_RR_UNKNOWN,
+			&livekit.ReconnectResponse{LastMessageSeq: 21},
+		))
+		require.True(t, p.IsReady())
+
+		require.Equal(t, 1, sink.WriteMessageCallCount())
+		res := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse)
+		require.NotNil(t, res.GetReconnect())
+		require.EqualValues(t, 21, res.GetReconnect().LastMessageSeq)
+	})
+
+	t.Run("the response goes out first", func(t *testing.T) {
+		p, sink := newMigratingParticipant()
+
+		// a room update while waiting is dropped, it is re-sent after the migration
+		require.NoError(t, p.SendRoomUpdate(&livekit.Room{Name: "test"}))
+		// participant updates while waiting are queued
+		require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{{
+			Sid:      "PA_other",
+			Identity: "other",
+			Version:  1,
+		}}))
+		require.Zero(t, sink.WriteMessageCallCount())
+
+		require.NoError(t, p.HandleReconnectAndSendResponse(
+			livekit.ReconnectReason_RR_UNKNOWN,
+			&livekit.ReconnectResponse{},
+		))
+
+		require.Equal(t, 2, sink.WriteMessageCallCount())
+		first := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse)
+		require.NotNil(t, first.GetReconnect())
+		second := sink.WriteMessageArgsForCall(1).(*livekit.SignalResponse)
+		require.NotNil(t, second.GetUpdate())
+		require.Len(t, second.GetUpdate().Participants, 1)
+		require.EqualValues(t, "PA_other", second.GetUpdate().Participants[0].Sid)
+	})
+
+	t.Run("a client that cannot handle the response is ready right away", func(t *testing.T) {
+		p := newParticipantForTestWithOpts("test", &participantOpts{
+			migration:       true,
+			protocolVersion: 17,
+			clientInfo:      &livekit.ClientInfo{Sdk: livekit.ClientInfo_JS, Version: "1.6.2"},
+		})
+		sink := p.params.Sink.(*routingfakes.FakeMessageSink)
+
+		require.NoError(t, p.HandleReconnectAndSendResponse(
+			livekit.ReconnectReason_RR_UNKNOWN,
+			&livekit.ReconnectResponse{},
+		))
+		require.True(t, p.IsReady())
+		require.Zero(t, sink.WriteMessageCallCount())
+
+		require.NoError(t, p.SendRoomUpdate(&livekit.Room{Name: "test"}))
+		require.Equal(t, 1, sink.WriteMessageCallCount())
+	})
+}
+
+func TestResumedParticipantWaitsForReconnectResponse(t *testing.T) {
+	// a resumed connection has to open with the ReconnectResponse too, and the
+	// participant is ready throughout, so the signaller holds messages back
+	newResumedParticipant := func(version string) (*ParticipantImpl, *routingfakes.FakeMessageSink) {
+		p := newParticipantForTestWithOpts("test", &participantOpts{
+			protocolVersion: 17,
+			clientInfo:      &livekit.ClientInfo{Sdk: livekit.ClientInfo_JS, Version: version},
+		})
+		require.True(t, p.IsReady())
+
+		sink := &routingfakes.FakeMessageSink{}
+		p.SwapResponseSink(sink, types.SignallingCloseReasonResume)
+		return p, sink
+	}
+
+	t.Run("the response goes out first", func(t *testing.T) {
+		p, sink := newResumedParticipant("2.15.2")
+
+		// a room update in this window is dropped, the resume path re-sends room state
+		require.NoError(t, p.SendRoomUpdate(&livekit.Room{Name: "test"}))
+		// participant updates are queued
+		require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{{
+			Sid:      "PA_other",
+			Identity: "other",
+			Version:  1,
+		}}))
+		require.Zero(t, sink.WriteMessageCallCount())
+
+		require.NoError(t, p.HandleReconnectAndSendResponse(
+			livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED,
+			&livekit.ReconnectResponse{LastMessageSeq: 7},
+		))
+
+		require.Equal(t, 2, sink.WriteMessageCallCount())
+		first := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse)
+		require.NotNil(t, first.GetReconnect())
+		require.EqualValues(t, 7, first.GetReconnect().LastMessageSeq)
+		second := sink.WriteMessageArgsForCall(1).(*livekit.SignalResponse)
+		require.NotNil(t, second.GetUpdate())
+		require.EqualValues(t, "PA_other", second.GetUpdate().Participants[0].Sid)
+	})
+
+	t.Run("a connection that opens without a response still delivers the queue", func(t *testing.T) {
+		p, sink := newResumedParticipant("2.15.2")
+
+		require.NoError(t, p.SendParticipantUpdate([]*livekit.ParticipantInfo{{
+			Sid:      "PA_other",
+			Identity: "other",
+			Version:  1,
+		}}))
+		require.Zero(t, sink.WriteMessageCallCount())
+
+		// no ReconnectResponse written, this is what the handshake window does on expiry
+		p.signaller.OpenHandshake()
+
+		require.Equal(t, 1, sink.WriteMessageCallCount())
+		res := sink.WriteMessageArgsForCall(0).(*livekit.SignalResponse)
+		require.NotNil(t, res.GetUpdate())
+		require.EqualValues(t, "PA_other", res.GetUpdate().Participants[0].Sid)
+	})
+
+	t.Run("a client that cannot handle the response is not held back", func(t *testing.T) {
+		p, sink := newResumedParticipant("1.6.2")
+
+		require.NoError(t, p.HandleReconnectAndSendResponse(
+			livekit.ReconnectReason_RR_SIGNAL_DISCONNECTED,
+			&livekit.ReconnectResponse{},
+		))
+		require.Zero(t, sink.WriteMessageCallCount())
+
+		require.NoError(t, p.SendRoomUpdate(&livekit.Room{Name: "test"}))
+		require.Equal(t, 1, sink.WriteMessageCallCount())
+	})
+}

pkg/rtc/participant_signal.go

@@ -81,7 +81,9 @@ func (p *ParticipantImpl) SendParticipantUpdate(participantsToUpdate []*livekit.
 		return nil
 	}
 
-	if !p.IsReady() {
+	// read under the lock the flush takes, so an update either queues before the flush
+	// or goes out after the connection has opened
+	if !p.IsReady() || p.signaller.HandshakePending() {
 		// queue up updates
 		p.queuedUpdates = append(p.queuedUpdates, participantsToUpdate...)
 		p.updateLock.Unlock()
@@ -177,11 +179,23 @@ func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit
 	p.TransportManager.HandleClientReconnect(reconnectReason)
 
 	if !p.params.ClientInfo.CanHandleReconnectResponse() {
-		return nil
+		// no ReconnectResponse opens this connection, so nothing to hold back
+		return p.reconnectResponseSentAndFlush()
 	}
-	if err := p.signaller.WriteMessage(p.signalling.SignalReconnectResponse(reconnectResponse)); err != nil {
+
+	// send reconnect response
+	err := p.signaller.WriteMessage(p.signalling.SignalReconnectResponse(reconnectResponse))
+
+	// mark sent after sending the message, so that nothing could slip through before
+	// ReconnectResponse is sent. Marked on a failed write too: the sink is gone in that
+	// case, and leaving it unmarked would hold back every message after it.
+	flushErr := p.reconnectResponseSentAndFlush()
+	if err != nil {
 		return err
 	}
+	if flushErr != nil {
+		return flushErr
+	}
 
 	if p.params.ProtocolVersion.SupportsDisconnectedUpdate() {
 		return p.sendDisconnectUpdatesForReconnect()
@@ -190,6 +204,44 @@ func (p *ParticipantImpl) HandleReconnectAndSendResponse(reconnectReason livekit
 	return nil
 }
 
+// reconnectResponseSentAndFlush makes a migrating in participant ready and sends what was
+// queued up while the connection was waiting for its ReconnectResponse.
+func (p *ParticipantImpl) reconnectResponseSentAndFlush() error {
+	// a successful write opens the connection by itself, this covers the paths that do
+	// not write one, i. e. a client that cannot handle it and a failed write
+	p.signaller.OpenHandshake()
+
+	p.updateLock.Lock()
+	p.reconnectResponseSent.Store(true)
+	queuedUpdates := p.queuedUpdates
+	p.queuedUpdates = nil
+	p.updateLock.Unlock()
+
+	if len(queuedUpdates) > 0 {
+		return p.SendParticipantUpdate(queuedUpdates)
+	}
+
+	return nil
+}
+
+// flushQueuedUpdates sends the updates queued while the connection was closed for the
+// handshake. Called when the connection opens, including when it opens without a
+// ReconnectResponse having been written.
+func (p *ParticipantImpl) flushQueuedUpdates() {
+	p.updateLock.Lock()
+	queuedUpdates := p.queuedUpdates
+	p.queuedUpdates = nil
+	p.updateLock.Unlock()
+
+	if len(queuedUpdates) == 0 {
+		return
+	}
+
+	if err := p.SendParticipantUpdate(queuedUpdates); err != nil {
+		p.params.Logger.Warnw("could not send queued participant updates", err)
+	}
+}
+
 func (p *ParticipantImpl) sendDisconnectUpdatesForReconnect() error {
 	lastSignalAt := p.TransportManager.LastSeenSignalAt()
 	var disconnectedParticipants []*livekit.ParticipantInfo

pkg/rtc/signalling/interfaces.go

@@ -32,6 +32,13 @@ type ParticipantSignaller interface {
 	GetResponseSink() routing.MessageSink
 	CloseSignalConnection(reason types.SignallingCloseReason)
 
+	// HandshakePending reports whether a resumed connection is still waiting for the
+	// ReconnectResponse it has to open with
+	HandshakePending() bool
+	// OpenHandshake lets messages flow on a resumed connection that does not open with
+	// a ReconnectResponse
+	OpenHandshake()
+
 	WriteMessage(msg proto.Message) error
 }
 

pkg/rtc/signalling/signallerasync.go

@@ -31,8 +31,9 @@ import (
 var _ ParticipantSignaller = (*signallerAsync)(nil)
 
 type SignallerAsyncParams struct {
-	Logger      logger.Logger
-	Participant types.LocalParticipant
+	Logger            logger.Logger
+	Participant       types.LocalParticipant
+	OnHandshakeOpened func()
 }
 
 type signallerAsync struct {
@@ -43,8 +44,11 @@ type signallerAsync struct {
 
 func NewSignallerAsync(params SignallerAsyncParams) ParticipantSignaller {
 	return &signallerAsync{
-		params:             params,
-		signallerAsyncBase: newSignallerAsyncBase(signallerAsyncBaseParams{Logger: params.Logger}),
+		params: params,
+		signallerAsyncBase: newSignallerAsyncBase(signallerAsyncBaseParams{
+			Logger:            params.Logger,
+			OnHandshakeOpened: params.OnHandshakeOpened,
+		}),
 	}
 }
 
@@ -69,17 +73,35 @@ func (s *signallerAsync) WriteMessage(msg proto.Message) error {
 		return nil
 	}
 
-	if !s.params.Participant.IsReady() {
-		if typed, ok := msg.(*livekit.SignalResponse); !ok {
-			s.params.Logger.Warnw(
-				"unknown message type", nil,
-				"messageType", fmt.Sprintf("%T", msg),
-			)
-		} else {
-			if typed.GetJoin() == nil {
-				return nil
+	// a signal connection opens with a join response, or with a reconnect response when
+	// it is resumed or migrated in. The client reads that first message as the handshake,
+	// so nothing may go out ahead of it.
+	isHandshake := false
+	isSdp := false
+	if typed, ok := msg.(*livekit.SignalResponse); !ok {
+		s.params.Logger.Warnw(
+			"unknown message type", nil,
+			"messageType", fmt.Sprintf("%T", msg),
+		)
+	} else {
+		isHandshake = typed.GetJoin() != nil || typed.GetReconnect() != nil
+		isSdp = typed.GetOffer() != nil || typed.GetAnswer() != nil
+	}
+
+	if !isHandshake && (!s.params.Participant.IsReady() || s.HandshakePending()) {
+		logFunc := s.params.Logger.Debugw
+		if isSdp {
+			// a dropped SDP leaves the negotiation waiting for the peer until the
+			// negotiation state machine recovers it, so make it visible
+			logFunc = func(msg string, keysAndValues ...any) {
+				s.params.Logger.Infow(msg, keysAndValues...)
 			}
 		}
+		logFunc(
+			"dropping message, connection has not opened yet",
+			"messageType", getMessageType(msg),
+		)
+		return nil
 	}
 
 	sink := s.GetResponseSink()
@@ -108,6 +130,10 @@ func (s *signallerAsync) WriteMessage(msg proto.Message) error {
 			return err
 		}
 	} else {
+		if isHandshake {
+			// the connection is open, hold nothing back any more
+			s.OpenHandshake()
+		}
 		s.params.Logger.Debugw("sent signal response", "response", logger.Proto(msg))
 	}
 	return nil

pkg/rtc/signalling/signallerasyncbase.go

@@ -16,6 +16,7 @@ package signalling
 
 import (
 	"sync"
+	"time"
 
 	"github.com/livekit/protocol/logger"
 
@@ -25,8 +26,16 @@ import (
 
 type signallerAsyncBaseParams struct {
 	Logger logger.Logger
+	// called when the connection opens, i. e. when messages held back for the
+	// handshake may be sent
+	OnHandshakeOpened func()
 }
 
+// how long a resumed connection holds messages back waiting for its ReconnectResponse.
+// A path that resumes without sending one lets messages flow after this instead of
+// holding them back for the rest of the session.
+const handshakeWindow = 5 * time.Second
+
 type signallerAsyncBase struct {
 	signallerUnimplemented
 
@@ -34,6 +43,13 @@ type signallerAsyncBase struct {
 
 	resSinkMu sync.Mutex
 	resSink   routing.MessageSink
+	// set while a resumed connection holds messages back waiting for its
+	// ReconnectResponse, the timer opens the connection if none is written.
+	// handshakeGeneration tells a timeout whether it belongs to the current wait, a
+	// Stop cannot cancel a callback that has already started running.
+	handshakePending    bool
+	handshakeGeneration uint32
+	handshakeTimer      *time.Timer
 }
 
 func newSignallerAsyncBase(params signallerAsyncBaseParams) *signallerAsyncBase {
@@ -46,8 +62,24 @@ func (s *signallerAsyncBase) SwapResponseSink(sink routing.MessageSink, reason t
 	s.resSinkMu.Lock()
 	oldSink := s.resSink
 	s.resSink = sink
+	// a resumed connection has to open with the ReconnectResponse, the client takes it
+	// only as the first message it reads
+	opened := false
+	switch {
+	case sink == nil:
+		// the connection is gone, keep anything queued for the next one
+		s.disarmHandshakeLocked()
+	case reason == types.SignallingCloseReasonResume:
+		s.armHandshakeLocked()
+	default:
+		opened = s.disarmHandshakeLocked()
+	}
 	s.resSinkMu.Unlock()
 
+	if opened {
+		s.notifyHandshakeOpened()
+	}
+
 	if oldSink != nil {
 		if sink != nil {
 			s.params.Logger.Debugw(
@@ -67,6 +99,80 @@ func (s *signallerAsyncBase) SwapResponseSink(sink routing.MessageSink, reason t
 	}
 }
 
+// HandshakePending is a plain read, so a caller can decide to hold a message back
+// while holding its own lock
+func (s *signallerAsyncBase) HandshakePending() bool {
+	s.resSinkMu.Lock()
+	defer s.resSinkMu.Unlock()
+
+	return s.handshakePending
+}
+
+func (s *signallerAsyncBase) OpenHandshake() {
+	s.resSinkMu.Lock()
+	opened := s.disarmHandshakeLocked()
+	s.resSinkMu.Unlock()
+
+	if opened {
+		s.notifyHandshakeOpened()
+	}
+}
+
+func (s *signallerAsyncBase) armHandshakeLocked() {
+	s.stopHandshakeTimerLocked()
+
+	s.handshakePending = true
+	s.handshakeGeneration++
+	generation := s.handshakeGeneration
+	// the connection opens on its own if nothing writes a ReconnectResponse, a path
+	// that resumes without one should not hold messages back for the whole session
+	s.handshakeTimer = time.AfterFunc(handshakeWindow, func() { s.onHandshakeTimeout(generation) })
+}
+
+// disarmHandshakeLocked reports whether it opened a connection that was holding
+// messages back
+func (s *signallerAsyncBase) disarmHandshakeLocked() bool {
+	s.stopHandshakeTimerLocked()
+
+	wasPending := s.handshakePending
+	s.handshakePending = false
+	// a timeout of the wait that just ended is stale
+	s.handshakeGeneration++
+	return wasPending
+}
+
+func (s *signallerAsyncBase) stopHandshakeTimerLocked() {
+	if s.handshakeTimer != nil {
+		s.handshakeTimer.Stop()
+		s.handshakeTimer = nil
+	}
+}
+
+func (s *signallerAsyncBase) onHandshakeTimeout(generation uint32) {
+	s.resSinkMu.Lock()
+	if generation != s.handshakeGeneration {
+		// the wait this timeout was armed for has ended, a later one may be in progress
+		s.resSinkMu.Unlock()
+		return
+	}
+	opened := s.disarmHandshakeLocked()
+	s.resSinkMu.Unlock()
+
+	if !opened {
+		return
+	}
+
+	s.params.Logger.Warnw("resumed connection did not open with a ReconnectResponse", nil)
+	s.notifyHandshakeOpened()
+}
+
+// notifyHandshakeOpened runs without resSinkMu held, the callback sends messages
+func (s *signallerAsyncBase) notifyHandshakeOpened() {
+	if s.params.OnHandshakeOpened != nil {
+		s.params.OnHandshakeOpened()
+	}
+}
+
 func (s *signallerAsyncBase) GetResponseSink() routing.MessageSink {
 	s.resSinkMu.Lock()
 	defer s.resSinkMu.Unlock()

pkg/rtc/signalling/signallerasyncbase_test.go

@@ -0,0 +1,157 @@
+package signalling
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+
+	"github.com/livekit/protocol/logger"
+
+	"github.com/livekit/livekit-server/pkg/routing/routingfakes"
+	"github.com/livekit/livekit-server/pkg/rtc/types"
+)
+
+func newTestSignallerBase(onHandshakeOpened func()) *signallerAsyncBase {
+	return newSignallerAsyncBase(signallerAsyncBaseParams{
+		Logger:            logger.GetLogger(),
+		OnHandshakeOpened: onHandshakeOpened,
+	})
+}
+
+// currentHandshakeGeneration is what a timeout armed right now would be tagged with
+func currentHandshakeGeneration(s *signallerAsyncBase) uint32 {
+	s.resSinkMu.Lock()
+	defer s.resSinkMu.Unlock()
+
+	return s.handshakeGeneration
+}
+
+func TestHandshakeGate(t *testing.T) {
+	t.Run("a resumed connection holds messages back", func(t *testing.T) {
+		s := newTestSignallerBase(nil)
+		require.False(t, s.HandshakePending())
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.True(t, s.HandshakePending())
+
+		s.OpenHandshake()
+		require.False(t, s.HandshakePending())
+	})
+
+	t.Run("other sink swaps do not hold messages back", func(t *testing.T) {
+		s := newTestSignallerBase(nil)
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonUnknown)
+		require.False(t, s.HandshakePending())
+	})
+
+	t.Run("closing the connection clears the gate", func(t *testing.T) {
+		s := newTestSignallerBase(nil)
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.True(t, s.HandshakePending())
+
+		s.CloseSignalConnection(types.SignallingCloseReasonParticipantClose)
+		require.False(t, s.HandshakePending())
+	})
+
+	t.Run("the handshake window opens the gate", func(t *testing.T) {
+		s := newTestSignallerBase(nil)
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.True(t, s.HandshakePending())
+
+		s.onHandshakeTimeout(currentHandshakeGeneration(s))
+		require.False(t, s.HandshakePending())
+	})
+
+	t.Run("a swap to a fresh connection opens the gate", func(t *testing.T) {
+		s := newTestSignallerBase(nil)
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.True(t, s.HandshakePending())
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonUnknown)
+		require.False(t, s.HandshakePending())
+	})
+
+	t.Run("a timeout of a wait that has ended does nothing", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		stale := currentHandshakeGeneration(s)
+
+		// the client resumes again while the first timeout is running, Stop cannot
+		// cancel a callback that has already started
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		s.onHandshakeTimeout(stale)
+
+		require.True(t, s.HandshakePending())
+		require.Zero(t, opened.Load())
+
+		// and the wait in progress still opens on its own timeout
+		s.onHandshakeTimeout(currentHandshakeGeneration(s))
+		require.False(t, s.HandshakePending())
+		require.EqualValues(t, 1, opened.Load())
+	})
+}
+
+func TestHandshakeGateNotifiesOnOpen(t *testing.T) {
+	t.Run("on an explicit open", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.Zero(t, opened.Load())
+
+		s.OpenHandshake()
+		require.EqualValues(t, 1, opened.Load())
+
+		// only the transition notifies
+		s.OpenHandshake()
+		require.EqualValues(t, 1, opened.Load())
+	})
+
+	t.Run("on the handshake window expiring", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		s.onHandshakeTimeout(currentHandshakeGeneration(s))
+
+		require.False(t, s.HandshakePending())
+		require.EqualValues(t, 1, opened.Load())
+	})
+
+	t.Run("the window fires without anything else touching the gate", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		require.Eventually(t, func() bool {
+			return !s.HandshakePending() && opened.Load() == 1
+		}, 2*handshakeWindow, 100*time.Millisecond)
+	})
+
+	t.Run("not on a connection close", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.SwapResponseSink(&routingfakes.FakeMessageSink{}, types.SignallingCloseReasonResume)
+		s.CloseSignalConnection(types.SignallingCloseReasonParticipantClose)
+
+		require.False(t, s.HandshakePending())
+		require.Zero(t, opened.Load())
+	})
+
+	t.Run("not when the gate was never armed", func(t *testing.T) {
+		var opened atomic.Int32
+		s := newTestSignallerBase(func() { opened.Inc() })
+
+		s.OpenHandshake()
+		require.Zero(t, opened.Load())
+	})
+}

pkg/rtc/signalling/signallerunimplemented.go

@@ -34,6 +34,12 @@ func (u *signallerUnimplemented) GetResponseSink() routing.MessageSink {
 
 func (u *signallerUnimplemented) CloseSignalConnection(reason types.SignallingCloseReason) {}
 
+func (u *signallerUnimplemented) HandshakePending() bool {
+	return false
+}
+
+func (u *signallerUnimplemented) OpenHandshake() {}
+
 func (u *signallerUnimplemented) WriteMessage(msg proto.Message) error {
 	return nil
 }