From 0aae7a4894006a37a535326d199835969e357780 Mon Sep 17 00:00:00 2001 From: Raja Subramanian Date: Tue, 1 Sep 2026 13:59:04 +0530 Subject: [PATCH] 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 * 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 * 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 * 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 * 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 --------- Co-authored-by: Claude Opus 5 --- pkg/rtc/participant.go | 14 +- pkg/rtc/participant_internal_test.go | 152 +++++++++++++++++ pkg/rtc/participant_signal.go | 58 ++++++- pkg/rtc/signalling/interfaces.go | 7 + pkg/rtc/signalling/signallerasync.go | 52 ++++-- pkg/rtc/signalling/signallerasyncbase.go | 106 ++++++++++++ pkg/rtc/signalling/signallerasyncbase_test.go | 157 ++++++++++++++++++ pkg/rtc/signalling/signallerunimplemented.go | 6 + 8 files changed, 532 insertions(+), 20 deletions(-) create mode 100644 pkg/rtc/signalling/signallerasyncbase_test.go diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index cdf7168de..17b07dac3 100644 --- a/pkg/rtc/participant.go +++ b/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, }) } diff --git a/pkg/rtc/participant_internal_test.go b/pkg/rtc/participant_internal_test.go index 2cde9181c..3a32808ac 100644 --- a/pkg/rtc/participant_internal_test.go +++ b/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()) + }) +} diff --git a/pkg/rtc/participant_signal.go b/pkg/rtc/participant_signal.go index 8f5c3dd2b..57f03f305 100644 --- a/pkg/rtc/participant_signal.go +++ b/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 diff --git a/pkg/rtc/signalling/interfaces.go b/pkg/rtc/signalling/interfaces.go index 99c3a0acc..2f8f59138 100644 --- a/pkg/rtc/signalling/interfaces.go +++ b/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 } diff --git a/pkg/rtc/signalling/signallerasync.go b/pkg/rtc/signalling/signallerasync.go index 0dc0853ee..c82dbc582 100644 --- a/pkg/rtc/signalling/signallerasync.go +++ b/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 diff --git a/pkg/rtc/signalling/signallerasyncbase.go b/pkg/rtc/signalling/signallerasyncbase.go index 0b4d2b665..28bfc968d 100644 --- a/pkg/rtc/signalling/signallerasyncbase.go +++ b/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() diff --git a/pkg/rtc/signalling/signallerasyncbase_test.go b/pkg/rtc/signalling/signallerasyncbase_test.go new file mode 100644 index 000000000..1d91a4c7e --- /dev/null +++ b/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()) + }) +} diff --git a/pkg/rtc/signalling/signallerunimplemented.go b/pkg/rtc/signalling/signallerunimplemented.go index 3592fd857..ae160a74f 100644 --- a/pkg/rtc/signalling/signallerunimplemented.go +++ b/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 }