diff --git a/engine.go b/engine.go index 7f994658..c013d411 100644 --- a/engine.go +++ b/engine.go @@ -77,6 +77,7 @@ type engineHandler interface { OnSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) OnSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate *livekit.SubscribedAudioCodecUpdate) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.MediaSectionsRequirement) + OnRequestResponse(requestResponse *livekit.RequestResponse) } // ------------------------------------------- @@ -130,6 +131,11 @@ type RTCEngine struct { trackPublishedListenersLock sync.Mutex trackPublishedListeners map[string]chan *livekit.TrackPublishedResponse + // signal requests waiting for a response, keyed by request id + pendingRequestsLock sync.Mutex + nextRequestID uint32 // guarded by pendingRequestsLock + pendingRequests map[uint32]chan proto.Message + subscriberPrimary bool hasPublish atomic.Bool closed atomic.Bool @@ -154,6 +160,7 @@ func NewRTCEngine( engineHandler: engineHandler, cbGetLocalParticipantSID: getLocalParticipantSID, trackPublishedListeners: make(map[string]chan *livekit.TrackPublishedResponse), + pendingRequests: make(map[uint32]chan proto.Message), reliableMsgSeq: 1, connectionManager: newConnectionManager(regionProvider), } @@ -359,6 +366,7 @@ func (e *RTCEngine) Close() { } e.connectionManager.setClosed() + e.abortPendingRequests() e.pclock.Lock() e.pendingPublisherOffer = webrtc.SessionDescription{} @@ -755,6 +763,76 @@ func (e *RTCEngine) UnregisterTrackPublishedListener(cid string) { e.trackPublishedListenersLock.Unlock() } +func (e *RTCEngine) newPendingRequest() *pendingRequest { + p := &pendingRequest{ + engine: e, + ch: make(chan proto.Message, 1), + } + + e.pendingRequestsLock.Lock() + p.id = e.allocateRequestIDLocked() + if e.closed.Load() { + // nothing will answer, fail Await right away + close(p.ch) + } else { + e.pendingRequests[p.id] = p.ch + } + e.pendingRequestsLock.Unlock() + + return p +} + +// allocateRequestIDLocked returns the next request id not in use by a pending request, skipping +// zero. Wrapping around is harmless: an id is reused only once its previous request is done. +func (e *RTCEngine) allocateRequestIDLocked() uint32 { + for { + e.nextRequestID++ + if e.nextRequestID == 0 { + continue + } + if _, inUse := e.pendingRequests[e.nextRequestID]; !inUse { + return e.nextRequestID + } + } +} + +func (e *RTCEngine) removePendingRequest(requestID uint32) { + e.pendingRequestsLock.Lock() + delete(e.pendingRequests, requestID) + e.pendingRequestsLock.Unlock() +} + +// deliverResponse hands res to the request waiting for requestID, if any. +func (e *RTCEngine) deliverResponse(requestID uint32, res proto.Message) bool { + if requestID == 0 { + return false + } + + e.pendingRequestsLock.Lock() + ch, ok := e.pendingRequests[requestID] + delete(e.pendingRequests, requestID) + e.pendingRequestsLock.Unlock() + if !ok { + return false + } + + // buffered, and removed from the map above so this is the only send + ch <- res + return true +} + +// abortPendingRequests fails every request still waiting for a response. +func (e *RTCEngine) abortPendingRequests() { + e.pendingRequestsLock.Lock() + pending := e.pendingRequests + e.pendingRequests = make(map[uint32]chan proto.Message) + e.pendingRequestsLock.Unlock() + + for _, ch := range pending { + close(ch) + } +} + func (e *RTCEngine) handleDataPacket(msg webrtc.DataChannelMessage) { packet, err := e.readDataPacket(msg) if err != nil { @@ -1043,6 +1121,8 @@ func (e *RTCEngine) cleanupConnection() { } func (e *RTCEngine) restartConnection() error { + // the new session will never answer requests made on the old one + e.abortPendingRequests() if err := e.join(nil, nil); err != nil { return err } @@ -1677,6 +1757,14 @@ func (e *RTCEngine) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit e.engineHandler.OnMediaSectionsRequirement(mediaSectionsRequirement) } +func (e *RTCEngine) OnRequestResponse(res *livekit.RequestResponse) { + if e.deliverResponse(res.GetRequestId(), res) { + return + } + + e.engineHandler.OnRequestResponse(res) +} + // ------------------------------------ func setConfiguration(pcTransport *PCTransport, configuration webrtc.Configuration) { diff --git a/requestresponse.go b/requestresponse.go new file mode 100644 index 00000000..e1e29983 --- /dev/null +++ b/requestresponse.go @@ -0,0 +1,97 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package lksdk + +import ( + "context" + "fmt" + "time" + + "github.com/livekit/protocol/livekit" + "google.golang.org/protobuf/proto" +) + +// SignalRequestError is returned when the server rejects a signal request. +type SignalRequestError struct { + RequestID uint32 + Reason livekit.RequestResponse_Reason + Message string +} + +func (e *SignalRequestError) Error() string { + return fmt.Sprintf("signal request rejected (%s): %s", e.Reason, e.Message) +} + +// pendingRequest reserves the response to a signal request that carries a request id: create it +// before sending the request, put ID() in the request_id field, then Await the response. The +// engine owns the registry these are tracked in, see (*RTCEngine).newPendingRequest. +type pendingRequest struct { + engine *RTCEngine + id uint32 + ch chan proto.Message +} + +func (p *pendingRequest) ID() uint32 { + return p.id +} + +// Await blocks until a response arrives, ctx is done, or the engine is closed. It has no timeout +// of its own, see withDefaultTimeout. A RequestResponse with a reason other than OK is returned +// as a *SignalRequestError. +func (p *pendingRequest) Await(ctx context.Context) (proto.Message, error) { + defer p.engine.removePendingRequest(p.id) + + select { + case res, ok := <-p.ch: + if !ok { + return nil, ErrAborted + } + if rr, isRequestResponse := res.(*livekit.RequestResponse); isRequestResponse && rr.GetReason() != livekit.RequestResponse_OK { + return nil, &SignalRequestError{ + RequestID: rr.GetRequestId(), + Reason: rr.GetReason(), + Message: rr.GetMessage(), + } + } + return res, nil + + case <-ctx.Done(): + return nil, ctx.Err() + } +} + +// awaitResponse is Await for requests whose success is reported by a dedicated message type. +func awaitResponse[T proto.Message](ctx context.Context, p *pendingRequest) (T, error) { + var zero T + + res, err := p.Await(ctx) + if err != nil { + return zero, err + } + typed, ok := res.(T) + if !ok { + return zero, fmt.Errorf("unexpected response %T to signal request %d", res, p.id) + } + return typed, nil +} + +// withDefaultTimeout applies deadline d to ctx when it has none. A deadline the caller set is +// never shortened. The returned cancel must be called. +func withDefaultTimeout(ctx context.Context, d time.Duration) (context.Context, context.CancelFunc) { + if _, ok := ctx.Deadline(); ok { + return ctx, func() {} + } + return context.WithTimeout(ctx, d) +} diff --git a/requestresponse_test.go b/requestresponse_test.go new file mode 100644 index 00000000..5135f85e --- /dev/null +++ b/requestresponse_test.go @@ -0,0 +1,210 @@ +// Copyright 2026 LiveKit, Inc. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package lksdk + +import ( + "context" + "math" + "testing" + "time" + + "github.com/livekit/protocol/livekit" + "github.com/stretchr/testify/require" +) + +func newTestEngine(t *testing.T) *RTCEngine { + engine := NewRoom(&RoomCallback{}).engine + t.Cleanup(engine.Close) + return engine +} + +func (e *RTCEngine) pendingRequestCount() int { + e.pendingRequestsLock.Lock() + defer e.pendingRequestsLock.Unlock() + return len(e.pendingRequests) +} + +func TestPendingRequestDelivered(t *testing.T) { + engine := newTestEngine(t) + + pending := engine.newPendingRequest() + require.NotZero(t, pending.ID()) + require.Equal(t, 1, engine.pendingRequestCount()) + + go engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()}) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + res, err := pending.Await(ctx) + require.NoError(t, err) + require.Equal(t, pending.ID(), res.(*livekit.RequestResponse).RequestId) + require.Zero(t, engine.pendingRequestCount()) +} + +func TestPendingRequestRejected(t *testing.T) { + engine := newTestEngine(t) + + pending := engine.newPendingRequest() + go engine.OnRequestResponse(&livekit.RequestResponse{ + RequestId: pending.ID(), + Reason: livekit.RequestResponse_NOT_ALLOWED, + Message: "no permission", + }) + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + _, err := pending.Await(ctx) + var rejected *SignalRequestError + require.ErrorAs(t, err, &rejected) + require.Equal(t, pending.ID(), rejected.RequestID) + require.Equal(t, livekit.RequestResponse_NOT_ALLOWED, rejected.Reason) + require.Equal(t, "no permission", rejected.Message) +} + +func TestPendingRequestTypedResponse(t *testing.T) { + engine := newTestEngine(t) + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + pending := engine.newPendingRequest() + go engine.deliverResponse(pending.ID(), &livekit.StoreDataBlobResponse{RequestId: pending.ID()}) + res, err := awaitResponse[*livekit.StoreDataBlobResponse](ctx, pending) + require.NoError(t, err) + require.Equal(t, pending.ID(), res.RequestId) + + pending = engine.newPendingRequest() + go engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID(), Reason: livekit.RequestResponse_NOT_FOUND}) + _, err = awaitResponse[*livekit.GetDataBlobResponse](ctx, pending) + var rejected *SignalRequestError + require.ErrorAs(t, err, &rejected) + require.Equal(t, livekit.RequestResponse_NOT_FOUND, rejected.Reason) + + // a response of the wrong type is an error, not a panic + pending = engine.newPendingRequest() + go engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID()}) + _, err = awaitResponse[*livekit.GetDataBlobResponse](ctx, pending) + require.Error(t, err) +} + +func TestPendingRequestContextDone(t *testing.T) { + engine := newTestEngine(t) + + pending := engine.newPendingRequest() + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err := pending.Await(ctx) + require.ErrorIs(t, err, context.Canceled) + require.Zero(t, engine.pendingRequestCount()) + + require.False(t, engine.deliverResponse(pending.ID(), &livekit.RequestResponse{RequestId: pending.ID()})) +} + +func TestWithDefaultTimeout(t *testing.T) { + ctx, cancel := withDefaultTimeout(context.Background(), time.Minute) + defer cancel() + deadline, ok := ctx.Deadline() + require.True(t, ok) + require.WithinDuration(t, time.Now().Add(time.Minute), deadline, 5*time.Second) + + // a caller's own deadline is left alone, even when it is longer than the default + parent, cancelParent := context.WithTimeout(context.Background(), time.Hour) + defer cancelParent() + parentDeadline, _ := parent.Deadline() + ctx, cancel = withDefaultTimeout(parent, time.Minute) + defer cancel() + deadline, ok = ctx.Deadline() + require.True(t, ok) + require.Equal(t, parentDeadline, deadline) + + engine := newTestEngine(t) + pending := engine.newPendingRequest() + ctx, cancel = withDefaultTimeout(context.Background(), 20*time.Millisecond) + defer cancel() + _, err := pending.Await(ctx) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Zero(t, engine.pendingRequestCount()) +} + +func TestPendingRequestAbortedOnClose(t *testing.T) { + engine := newTestEngine(t) + + pending := engine.newPendingRequest() + engine.Close() + _, err := pending.Await(context.Background()) + require.ErrorIs(t, err, ErrAborted) + + _, err = engine.newPendingRequest().Await(context.Background()) + require.ErrorIs(t, err, ErrAborted) + require.Zero(t, engine.pendingRequestCount()) +} + +func TestUncorrelatedResponseIgnored(t *testing.T) { + engine := newTestEngine(t) + + // request id 0 is never matched to a pending request + require.False(t, engine.deliverResponse(0, &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED})) + require.False(t, engine.deliverResponse(12345, &livekit.RequestResponse{RequestId: 12345})) +} + +func TestRequestIDAllocation(t *testing.T) { + engine := newTestEngine(t) + + seen := make(map[uint32]struct{}) + for range 100 { + id := engine.newPendingRequest().ID() + require.NotZero(t, id) + _, dup := seen[id] + require.False(t, dup) + seen[id] = struct{}{} + } + + // the counter wraps around without handing out zero or an id that is still in use + engine.pendingRequestsLock.Lock() + engine.nextRequestID = math.MaxUint32 - 1 + engine.pendingRequestsLock.Unlock() + require.Equal(t, uint32(math.MaxUint32), engine.newPendingRequest().ID()) + require.Equal(t, uint32(101), engine.newPendingRequest().ID()) +} + +type requestResponseRecorder struct { + *Room + responses chan *livekit.RequestResponse +} + +func (r *requestResponseRecorder) OnRequestResponse(res *livekit.RequestResponse) { + r.responses <- res +} + +func TestRequestResponseForwardedToHandler(t *testing.T) { + recorder := &requestResponseRecorder{ + Room: NewRoom(&RoomCallback{}), + responses: make(chan *livekit.RequestResponse, 1), + } + t.Cleanup(recorder.Room.engine.Close) + engine := NewRTCEngine(false, recorder, func() string { return "" }, newRegionURLProvider()) + t.Cleanup(engine.Close) + + // no request id: nothing to correlate, handed to the handler + uncorrelated := &livekit.RequestResponse{Reason: livekit.RequestResponse_NOT_ALLOWED} + engine.OnRequestResponse(uncorrelated) + require.Same(t, uncorrelated, <-recorder.responses) + + // a waiting request consumes its response, the handler is not involved + pending := engine.newPendingRequest() + engine.OnRequestResponse(&livekit.RequestResponse{RequestId: pending.ID()}) + _, err := pending.Await(context.Background()) + require.NoError(t, err) + require.Empty(t, recorder.responses) +} diff --git a/room.go b/room.go index bfa3039f..075f17f0 100644 --- a/room.go +++ b/room.go @@ -1399,6 +1399,18 @@ func (r *Room) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.Medi publisher.Negotiate() } +// OnRequestResponse handles responses to signal requests that no caller is waiting on. +func (r *Room) OnRequestResponse(res *livekit.RequestResponse) { + if res.GetReason() != livekit.RequestResponse_OK { + r.log.Warnw( + "signal request failed", nil, + "requestID", res.GetRequestId(), + "reason", res.GetReason(), + "message", res.GetMessage(), + ) + } +} + func (r *Room) OnStreamHeader(streamHeader *livekit.DataStream_Header, participantIdentity string) { switch header := streamHeader.ContentHeader.(type) { case *livekit.DataStream_Header_TextHeader: diff --git a/signalling/interfaces.go b/signalling/interfaces.go index 526f7d0b..ec57eb95 100644 --- a/signalling/interfaces.go +++ b/signalling/interfaces.go @@ -181,4 +181,5 @@ type SignalProcessor interface { OnSubscribedQualityUpdate(subscribedQualityUpdate *livekit.SubscribedQualityUpdate) OnSubscribedAudioCodecUpdate(subscribedAudioCodecUpdate *livekit.SubscribedAudioCodecUpdate) OnMediaSectionsRequirement(mediaSectionsRequirement *livekit.MediaSectionsRequirement) + OnRequestResponse(requestResponse *livekit.RequestResponse) } diff --git a/signalling/signalhandler.go b/signalling/signalhandler.go index 74bd26e0..638b85a2 100644 --- a/signalling/signalhandler.go +++ b/signalling/signalhandler.go @@ -121,6 +121,9 @@ func (s *signalhandler) HandleMessage(msg proto.Message) error { case *livekit.SignalResponse_MediaSectionsRequirement: s.params.Processor.OnMediaSectionsRequirement(payload.MediaSectionsRequirement) + + case *livekit.SignalResponse_RequestResponse: + s.params.Processor.OnRequestResponse(payload.RequestResponse) } return nil